swoft support

classes/fpf/thrift/SwoftHttpClient.php

+<?php
+
+/**
+ * Created by PhpStorm.
+ * User: rlgyzhcn
+ * Date: 18-7-9
+ * Time: 下午6:50
+ */
+
+namespace fpf\thrift;
+
+use Thrift\Exception\TTransportException;
+use Thrift\Factory\TStringFuncFactory;
+use Swoft\HttpClient\Client;
+
+class SwoftHttpClient extends \Thrift\Transport\TTransport
+{
+    /**
+     * The host to connect to
+     *
+     * @var string
+     */
+    protected $host_;
+
+    /**
+     * The port to connect on
+     *
+     * @var int
+     */
+    protected $port_;
+
+    /**
+     * The URI to request
+     *
+     * @var string
+     */
+    protected $uri_;
+
+    /**
+     * The scheme to use for the request, i.e. http, https
+     *
+     * @var string
+     */
+    protected $scheme_;
+
+    /**
+     * Buffer for the HTTP request data
+     *
+     * @var string
+     */
+    protected $request_;
+
+    /**
+     * Buffer for the HTTP response data.
+     *
+     * @var binary string
+     */
+    protected $response_;
+
+    /**
+     * Read timeout
+     *
+     * @var float
+     */
+    protected $timeout_;
+
+    /**
+     * http headers
+     *
+     * @var array
+     */
+    protected $headers_;
+
+    protected $uri_no_func;
+
+    /**
+     * http connect timeout
+     *
+     * @var int
+     */
+    protected $connect_timeout_ms;
+
+    /**
+     * http timeout
+     *
+     * @var int
+     */
+    protected $timeout_ms;
+
+    /**
+     * Http 客户端
+     *
+     * @var Client
+     */
+    private static $handle;
+
+    /**
+     * Make a new HTTP client.
+     *
+     * @param string $host
+     * @param int $port
+     * @param string $uri_no_func
+     */
+    public function __construct($host, $port = 80, $uri_no_func = '', $scheme = 'http', $config)
+    {
+        if ((TStringFuncFactory::create()->strlen($uri_no_func) > 0) && ($uri_no_func {
+            0} != '/')) {
+            $uri_no_func = '/' . $uri_no_func;
+        }
+        $this->scheme_ = $scheme;
+        $this->host_ = $host;
+        $this->port_ = $port;
+        $this->uri_no_func = $uri_no_func;
+        $this->uri_ = $uri_no_func;
+        $this->request_ = '';
+        $this->response_ = null;
+        $this->timeout_ = null;
+        $this->connect_timeout_ms = $config['connect_timeout'] ?? 10;
+        $this->timeout_ms = $config['timeout'] ?? 100;
+        $this->headers_ = array();
+    }
+
+    public function setFuncUri($uri_func)
+    {
+        $this->uri_ = rtrim($this->uri_no_func, '/') . '/' . $uri_func;
+    }
+
+    /**
+     * Set read timeout
+     *
+     * @param float $timeout
+     */
+    public function setTimeoutSecs($timeout)
+    {
+        $this->timeout_ = $timeout;
+    }
+
+    /**
+     * Whether this transport is open.
+     *
+     * @return boolean true if open
+     */
+    public function isOpen()
+    {
+        return true;
+    }
+
+    /**
+     * Open the transport for reading/writing
+     *
+     * @throws TTransportException if cannot open
+     */
+    public function open()
+    {
+    }
+
+    /**
+     * Close the transport.
+     */
+    public function close()
+    {
+        $this->request_ = '';
+        $this->response_ = null;
+    }
+
+    /**
+     * Read some data into the array.
+     *
+     * @param int $len How much to read
+     * @return string The data that has been read
+     * @throws TTransportException if cannot read any more data
+     */
+    public function read($len)
+    {
+        if ($len >= strlen($this->response_)) {
+            return $this->response_;
+        } else {
+            $ret = substr($this->response_, 0, $len);
+            $this->response_ = substr($this->response_, $len);
+
+            return $ret;
+        }
+    }
+
+    /**
+     * Writes some data into the pending buffer
+     *
+     * @param string $buf The data to write
+     * @throws TTransportException if writing fails
+     */
+    public function write($buf)
+    {
+        $this->request_ .= $buf;
+    }
+
+    /**
+     * Opens and sends the actual request over the HTTP connection
+     *
+     * @throws TTransportException if a writing error occurs
+     */
+    public function flush()
+    {
+        if (!self::$handle) {
+            register_shutdown_function(array('Thrift\\Transport\\TCurlClient', 'closeCurlHandle'));
+            self::$handle = new Client(['adapter' => 'co']);//使用协程驱动
+        }
+        $options = [];
+        $host = $this->host_ . ($this->port_ != 80 ? ':' . $this->port_ : '');
+        $defaultHeaders = array(
+            'Accept' => 'application/x-thrift',
+            'Content-Type' => 'application/x-thrift',
+            'User-Agent' => 'PHP/TCurlClient',
+            'Content-Length' => TStringFuncFactory::create()->strlen($this->request_)
+        );
+        $options['base_uri'] = $this->scheme_ . "://" . $host;
+        $options['headers'] = array_merge($defaultHeaders, $this->headers_);
+        if (defined('VERSION_KEY') && defined('VERSION_VALUE_NO_PTEFIX')) {
+            $options['headers'][VERSION_KEY] = VERSION_VALUE_NO_PTEFIX;
+        }
+        $options['body'] = $this->request_;
+        $options['timeout'] = $this->timeout_;
+        $this->request_ = '';
+        $this->response_ = self::$handle->request('POST', $this->uri_, $options)->getResult();
+        // Connect failed?
+        if (!$this->response_) {
+            $error = 'SwoftHttpClient: Could not connect to ' . $this->scheme_ . "://" . $host . $this->uri_;
+            throw new TTransportException($error, TTransportException::NOT_OPEN);
+        }
+    }
+
+    public static function closeCurlHandle()
+    {
+        if (self::$handle) {
+            self::$handle = null;
+        }
+    }
+
+    public function addHeaders($headers)
+    {
+        $this->headers_ = array_merge($this->headers_, $headers);
+    }
+}
\ No newline at end of file

classes/fpf/thrift/ThriftServiceFactoryProxy.php

@@ -47,14 +47,25 @@ class ThriftServiceFactoryProxy
             self::BINARY => 'application/thrift-binary',
             self::JSON => 'application/thrift-json',
         ];
-        $this->socket = new TCurlClient(
-            $service_real['host'],
-            $service_real['port'],
-            '/' . $service_name_space_str . '/' . $service_str . '/',
-            'http',
-            $service_real
-        );
-        $this->socket->addHeaders(['Content-type' => $content_type_mapping[$content_type]]);
+        if (defined('IS_SWOOLE_SERVICE') && IS_SWOOLE_SERVICE === true) {
+            $this->socket = new SwoftHttpClient(
+                $service_real['host'],
+                $service_real['port'],
+                '/' . $service_name_space_str . '/' . $service_str . '/',
+                'http',
+                $service_real
+            );
+            $this->socket->addHeaders(['Contents-Type' => $content_type_mapping[$content_type]]);
+        } else {
+            $this->socket = new TCurlClient(
+                $service_real['host'],
+                $service_real['port'],
+                '/' . $service_name_space_str . '/' . $service_str . '/',
+                'http',
+                $service_real
+            );
+            $this->socket->addHeaders(['Content-type' => $content_type_mapping[$content_type]]);
+        }
         $this->transport = new \Thrift\Transport\TBufferedTransport($this->socket, 1024, 1024);
         if (self::BINARY === $content_type) {
             $protocol = new \Thrift\Protocol\TBinaryProtocol($this->transport, $strict_read = true, $strict_write = true);