diff options
Diffstat (limited to 'vendor/ruflin/elastica/lib/Elastica/Transport')
9 files changed, 1009 insertions, 0 deletions
diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/AbstractTransport.php b/vendor/ruflin/elastica/lib/Elastica/Transport/AbstractTransport.php new file mode 100644 index 00000000..c40b5107 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/AbstractTransport.php @@ -0,0 +1,113 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Connection; +use Elastica\Exception\InvalidException; +use Elastica\Param; +use Elastica\Request; + +/** + * Elastica Abstract Transport object. + * + * @author Nicolas Ruflin <spam@ruflin.com> + */ +abstract class AbstractTransport extends Param +{ + /** + * @var \Elastica\Connection + */ + protected $_connection; + + /** + * Construct transport. + * + * @param \Elastica\Connection $connection Connection object + */ + public function __construct(Connection $connection = null) + { + if ($connection) { + $this->setConnection($connection); + } + } + + /** + * @return \Elastica\Connection Connection object + */ + public function getConnection() + { + return $this->_connection; + } + + /** + * @param \Elastica\Connection $connection Connection object + * + * @return $this + */ + public function setConnection(Connection $connection) + { + $this->_connection = $connection; + + return $this; + } + + /** + * Executes the transport request. + * + * @param \Elastica\Request $request Request object + * @param array $params Hostname, port, path, ... + * + * @return \Elastica\Response Response object + */ + abstract public function exec(Request $request, array $params); + + /** + * Create a transport. + * + * The $transport parameter can be one of the following values: + * + * * string: The short name of a transport. For instance "Http", "Memcache" or "Thrift" + * * object: An already instantiated instance of a transport + * * array: An array with a "type" key which must be set to one of the two options. All other + * keys in the array will be set as parameters in the transport instance + * + * @param mixed $transport A transport definition + * @param \Elastica\Connection $connection A connection instance + * @param array $params Parameters for the transport class + * + * @throws \Elastica\Exception\InvalidException + * + * @return AbstractTransport + */ + public static function create($transport, Connection $connection, array $params = array()) + { + if (is_array($transport) && isset($transport['type'])) { + $transportParams = $transport; + unset($transportParams['type']); + + $params = array_replace($params, $transportParams); + $transport = $transport['type']; + } + + if (is_string($transport)) { + $className = 'Elastica\\Transport\\'.$transport; + + if (!class_exists($className)) { + throw new InvalidException('Invalid transport'); + } + + $transport = new $className(); + } + + if ($transport instanceof self) { + $transport->setConnection($connection); + + foreach ($params as $key => $value) { + $transport->setParam($key, $value); + } + } else { + throw new InvalidException('Invalid transport'); + } + + return $transport; + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Guzzle.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Guzzle.php new file mode 100644 index 00000000..5c98d83b --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Guzzle.php @@ -0,0 +1,179 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Connection; +use Elastica\Exception\Connection\GuzzleException; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; +use GuzzleHttp\Client; +use GuzzleHttp\Exception\TransferException; +use GuzzleHttp\Stream\Stream; + +/** + * Elastica Guzzle Transport object. + * + * @author Milan Magudia <milan@magudia.com> + */ +class Guzzle extends AbstractTransport +{ + /** + * Http scheme. + * + * @var string Http scheme + */ + protected $_scheme = 'http'; + + /** + * Curl resource to reuse. + * + * @var resource Guzzle resource to reuse + */ + protected static $_guzzleClientConnection = null; + + /** + * Makes calls to the elasticsearch server. + * + * All calls that are made to the server are done through this function + * + * @param \Elastica\Request $request + * @param array $params Host, Port, ... + * + * @throws \Elastica\Exception\ConnectionException + * @throws \Elastica\Exception\ResponseException + * @throws \Elastica\Exception\Connection\HttpException + * + * @return \Elastica\Response Response object + */ + public function exec(Request $request, array $params) + { + $connection = $this->getConnection(); + + $client = $this->_getGuzzleClient($this->_getBaseUrl($connection), $connection->isPersistent()); + + $options = array( + 'exceptions' => false, // 4xx and 5xx is expected and NOT an exceptions in this context + ); + if ($connection->getTimeout()) { + $options['timeout'] = $connection->getTimeout(); + } + + $proxy = $connection->getProxy(); + + // See: https://github.com/facebook/hhvm/issues/4875 + if (is_null($proxy) && defined('HHVM_VERSION')) { + $proxy = getenv('http_proxy') ?: null; + } + + if (!is_null($proxy)) { + $options['proxy'] = $proxy; + } + + $req = $client->createRequest($request->getMethod(), $this->_getActionPath($request), $options); + $req->setHeaders($connection->hasConfig('headers') ? $connection->getConfig('headers') : array()); + + $data = $request->getData(); + if (!empty($data) || '0' === $data) { + if ($req->getMethod() == Request::GET) { + $req->setMethod(Request::POST); + } + + if ($this->hasParam('postWithRequestBody') && $this->getParam('postWithRequestBody') == true) { + $request->setMethod(Request::POST); + $req->setMethod(Request::POST); + } + + if (is_array($data)) { + $content = JSON::stringify($data, 'JSON_ELASTICSEARCH'); + } else { + $content = $data; + } + $req->setBody(Stream::factory($content)); + } + + try { + $start = microtime(true); + $res = $client->send($req); + $end = microtime(true); + } catch (TransferException $ex) { + throw new GuzzleException($ex, $request, new Response($ex->getMessage())); + } + + $response = new Response((string) $res->getBody(), $res->getStatusCode()); + $response->setQueryTime($end - $start); + + $response->setTransferInfo( + array( + 'request_header' => $request->getMethod(), + 'http_code' => $res->getStatusCode(), + ) + ); + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + return $response; + } + + /** + * Return Guzzle resource. + * + * @param bool $persistent False if not persistent connection + * + * @return resource Connection resource + */ + protected function _getGuzzleClient($baseUrl, $persistent = true) + { + if (!$persistent || !self::$_guzzleClientConnection) { + self::$_guzzleClientConnection = new Client(array('base_url' => $baseUrl)); + } + + return self::$_guzzleClientConnection; + } + + /** + * Builds the base url for the guzzle connection. + * + * @param \Elastica\Connection $connection + */ + protected function _getBaseUrl(Connection $connection) + { + // If url is set, url is taken. Otherwise port, host and path + $url = $connection->hasConfig('url') ? $connection->getConfig('url') : ''; + + if (!empty($url)) { + $baseUri = $url; + } else { + $baseUri = $this->_scheme.'://'.$connection->getHost().':'.$connection->getPort().'/'.$connection->getPath(); + } + + return rtrim($baseUri, '/'); + } + + /** + * Builds the action path url for each request. + * + * @param \Elastica\Request $request + */ + protected function _getActionPath(Request $request) + { + $action = $request->getPath(); + if ($action) { + $action = '/'.ltrim($action, '/'); + } + $query = $request->getQuery(); + + if (!empty($query)) { + $action .= '?'.http_build_query($query); + } + + return $action; + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Http.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Http.php new file mode 100644 index 00000000..1a33c0a6 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Http.php @@ -0,0 +1,190 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Exception\Connection\HttpException; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; + +/** + * Elastica Http Transport object. + * + * @author Nicolas Ruflin <spam@ruflin.com> + */ +class Http extends AbstractTransport +{ + /** + * Http scheme. + * + * @var string Http scheme + */ + protected $_scheme = 'http'; + + /** + * Curl resource to reuse. + * + * @var resource Curl resource to reuse + */ + protected static $_curlConnection = null; + + /** + * Makes calls to the elasticsearch server. + * + * All calls that are made to the server are done through this function + * + * @param \Elastica\Request $request + * @param array $params Host, Port, ... + * + * @throws \Elastica\Exception\ConnectionException + * @throws \Elastica\Exception\ResponseException + * @throws \Elastica\Exception\Connection\HttpException + * + * @return \Elastica\Response Response object + */ + public function exec(Request $request, array $params) + { + $connection = $this->getConnection(); + + $conn = $this->_getConnection($connection->isPersistent()); + + // If url is set, url is taken. Otherwise port, host and path + $url = $connection->hasConfig('url') ? $connection->getConfig('url') : ''; + + if (!empty($url)) { + $baseUri = $url; + } else { + $baseUri = $this->_scheme.'://'.$connection->getHost().':'.$connection->getPort().'/'.$connection->getPath(); + } + + $baseUri .= $request->getPath(); + + $query = $request->getQuery(); + + if (!empty($query)) { + $baseUri .= '?'.http_build_query($query); + } + + curl_setopt($conn, CURLOPT_URL, $baseUri); + curl_setopt($conn, CURLOPT_TIMEOUT, $connection->getTimeout()); + curl_setopt($conn, CURLOPT_FORBID_REUSE, 0); + + /* @see Connection::setConnectTimeout() */ + $connectTimeout = $connection->getConnectTimeout(); + if ($connectTimeout > 0) { + curl_setopt($conn, CURLOPT_CONNECTTIMEOUT, $connectTimeout); + } + + $proxy = $connection->getProxy(); + + // See: https://github.com/facebook/hhvm/issues/4875 + if (is_null($proxy) && defined('HHVM_VERSION')) { + $proxy = getenv('http_proxy') ?: null; + } + + if (!is_null($proxy)) { + curl_setopt($conn, CURLOPT_PROXY, $proxy); + } + + $this->_setupCurl($conn); + + $headersConfig = $connection->hasConfig('headers') ? $connection->getConfig('headers') : array(); + + if (!empty($headersConfig)) { + $headers = array(); + while (list($header, $headerValue) = each($headersConfig)) { + array_push($headers, $header.': '.$headerValue); + } + + curl_setopt($conn, CURLOPT_HTTPHEADER, $headers); + } + + // TODO: REFACTOR + $data = $request->getData(); + $httpMethod = $request->getMethod(); + + if (!empty($data) || '0' === $data) { + if ($this->hasParam('postWithRequestBody') && $this->getParam('postWithRequestBody') == true) { + $httpMethod = Request::POST; + } + + if (is_array($data)) { + $content = JSON::stringify($data, 'JSON_ELASTICSEARCH'); + } else { + $content = $data; + } + + // Escaping of / not necessary. Causes problems in base64 encoding of files + $content = str_replace('\/', '/', $content); + + curl_setopt($conn, CURLOPT_POSTFIELDS, $content); + } else { + curl_setopt($conn, CURLOPT_POSTFIELDS, ''); + } + + curl_setopt($conn, CURLOPT_NOBODY, $httpMethod == 'HEAD'); + + curl_setopt($conn, CURLOPT_CUSTOMREQUEST, $httpMethod); + + $start = microtime(true); + + // cURL opt returntransfer leaks memory, therefore OB instead. + ob_start(); + curl_exec($conn); + $responseString = ob_get_clean(); + + $end = microtime(true); + + // Checks if error exists + $errorNumber = curl_errno($conn); + + $response = new Response($responseString, curl_getinfo($conn, CURLINFO_HTTP_CODE)); + $response->setQueryTime($end - $start); + $response->setTransferInfo(curl_getinfo($conn)); + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + if ($errorNumber > 0) { + throw new HttpException($errorNumber, $request, $response); + } + + return $response; + } + + /** + * Called to add additional curl params. + * + * @param resource $curlConnection Curl connection + */ + protected function _setupCurl($curlConnection) + { + if ($this->getConnection()->hasConfig('curl')) { + foreach ($this->getConnection()->getConfig('curl') as $key => $param) { + curl_setopt($curlConnection, $key, $param); + } + } + } + + /** + * Return Curl resource. + * + * @param bool $persistent False if not persistent connection + * + * @return resource Connection resource + */ + protected function _getConnection($persistent = true) + { + if (!$persistent || !self::$_curlConnection) { + self::$_curlConnection = curl_init(); + } + + return self::$_curlConnection; + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/HttpAdapter.php b/vendor/ruflin/elastica/lib/Elastica/Transport/HttpAdapter.php new file mode 100644 index 00000000..efc27ab5 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/HttpAdapter.php @@ -0,0 +1,156 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Connection; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\JSON; +use Elastica\Request as ElasticaRequest; +use Elastica\Response as ElasticaResponse; +use Ivory\HttpAdapter\HttpAdapterInterface; +use Ivory\HttpAdapter\Message\Request as HttpAdapterRequest; +use Ivory\HttpAdapter\Message\Response as HttpAdapterResponse; +use Ivory\HttpAdapter\Message\Stream\StringStream; + +class HttpAdapter extends AbstractTransport +{ + /** + * @var HttpAdapterInterface + */ + private $httpAdapter; + + /** + * @var string + */ + private $_scheme = 'http'; + + /** + * Construct transport. + */ + public function __construct(Connection $connection = null, HttpAdapterInterface $httpAdapter) + { + parent::__construct($connection); + $this->httpAdapter = $httpAdapter; + } + + /** + * Makes calls to the elasticsearch server. + * + * All calls that are made to the server are done through this function + * + * @param \Elastica\Request $elasticaRequest + * @param array $params Host, Port, ... + * + * @throws \Elastica\Exception\ConnectionException + * @throws \Elastica\Exception\ResponseException + * @throws \Elastica\Exception\Connection\HttpException + * + * @return \Elastica\Response Response object + */ + public function exec(ElasticaRequest $elasticaRequest, array $params) + { + $connection = $this->getConnection(); + + if ($timeout = $connection->getTimeout()) { + $this->httpAdapter->getConfiguration()->setTimeout($timeout); + } + + $httpAdapterRequest = $this->_createHttpAdapterRequest($elasticaRequest, $connection); + + $start = microtime(true); + $httpAdapterResponse = $this->httpAdapter->sendRequest($httpAdapterRequest); + $end = microtime(true); + + $elasticaResponse = $this->_createElasticaResponse($httpAdapterResponse, $connection); + $elasticaResponse->setQueryTime($end - $start); + + $elasticaResponse->setTransferInfo( + array( + 'request_header' => $httpAdapterRequest->getMethod(), + 'http_code' => $httpAdapterResponse->getStatusCode(), + ) + ); + + if ($elasticaResponse->hasError()) { + throw new ResponseException($elasticaRequest, $elasticaResponse); + } + + if ($elasticaResponse->hasFailedShards()) { + throw new PartialShardFailureException($elasticaRequest, $elasticaResponse); + } + + return $elasticaResponse; + } + + /** + * @param HttpAdapterResponse $httpAdapterResponse + * + * @return ElasticaResponse + */ + protected function _createElasticaResponse(HttpAdapterResponse $httpAdapterResponse) + { + return new ElasticaResponse((string) $httpAdapterResponse->getBody(), $httpAdapterResponse->getStatusCode()); + } + + /** + * @param ElasticaRequest $elasticaRequest + * @param Connection $connection + * + * @return HttpAdapterRequest + */ + protected function _createHttpAdapterRequest(ElasticaRequest $elasticaRequest, Connection $connection) + { + $data = $elasticaRequest->getData(); + $body = null; + $method = $elasticaRequest->getMethod(); + $headers = $connection->hasConfig('headers') ?: array(); + if (!empty($data) || '0' === $data) { + if ($method == ElasticaRequest::GET) { + $method = ElasticaRequest::POST; + } + + if ($this->hasParam('postWithRequestBody') && $this->getParam('postWithRequestBody') == true) { + $elasticaRequest->setMethod(ElasticaRequest::POST); + $method = ElasticaRequest::POST; + } + + if (is_array($data)) { + $body = JSON::stringify($data, 'JSON_ELASTICSEARCH'); + } else { + $body = $data; + } + } + + $url = $this->_getUri($elasticaRequest, $connection); + $streamBody = new StringStream($body); + + return new HttpAdapterRequest($url, $method, HttpAdapterRequest::PROTOCOL_VERSION_1_1, $headers, $streamBody); + } + + /** + * @param ElasticaRequest $request + * @param \Elastica\Connection $connection + * + * @return string + */ + protected function _getUri(ElasticaRequest $request, Connection $connection) + { + $url = $connection->hasConfig('url') ? $connection->getConfig('url') : ''; + + if (!empty($url)) { + $baseUri = $url; + } else { + $baseUri = $this->_scheme.'://'.$connection->getHost().':'.$connection->getPort().'/'.$connection->getPath(); + } + + $baseUri .= $request->getPath(); + + $query = $request->getQuery(); + + if (!empty($query)) { + $baseUri .= '?'.http_build_query($query); + } + + return $baseUri; + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Https.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Https.php new file mode 100644 index 00000000..b2b489dd --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Https.php @@ -0,0 +1,27 @@ +<?php +namespace Elastica\Transport; + +/** + * Elastica Http Transport object. + * + * @author Nicolas Ruflin <spam@ruflin.com> + */ +class Https extends Http +{ + /** + * Https scheme. + * + * @var string https scheme + */ + protected $_scheme = 'https'; + + /** + * Overloads setupCurl to set SSL params. + * + * @param resource $connection Curl connection resource + */ + protected function _setupCurl($connection) + { + parent::_setupCurl($connection); + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Memcache.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Memcache.php new file mode 100644 index 00000000..fb56cdf4 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Memcache.php @@ -0,0 +1,109 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Exception\Connection\MemcacheException; +use Elastica\Exception\InvalidException; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; + +/** + * Elastica Memcache Transport object. + * + * @author Nicolas Ruflin <spam@ruflin.com> + * + * @deprecated The memcached transport is deprecated as of ES 1.5, and will be removed in ES 2.0 + */ +class Memcache extends AbstractTransport +{ + const MAX_KEY_LENGTH = 250; + + /** + * Makes calls to the elasticsearch server. + * + * @param \Elastica\Request $request + * @param array $params Host, Port, ... + * + * @throws \Elastica\Exception\ResponseException + * @throws \Elastica\Exception\InvalidException + * + * @return \Elastica\Response Response object + */ + public function exec(Request $request, array $params) + { + $memcache = new \Memcache(); + $memcache->connect($this->getConnection()->getHost(), $this->getConnection()->getPort()); + + $data = $request->getData(); + + $content = ''; + + if (!empty($data) || '0' === $data) { + if (is_array($data)) { + $content = JSON::stringify($data); + } else { + $content = $data; + } + + // Escaping of / not necessary. Causes problems in base64 encoding of files + $content = str_replace('\/', '/', $content); + } + + $responseString = ''; + + $start = microtime(true); + + switch ($request->getMethod()) { + case Request::POST: + case Request::PUT: + $key = $request->getPath(); + $this->_checkKeyLength($key); + $memcache->set($key, $content); + break; + case Request::GET: + $key = $request->getPath().'?source='.$content; + $this->_checkKeyLength($key); + $responseString = $memcache->get($key); + break; + case Request::DELETE: + $key = $request->getPath().'?source='.$content; + $this->_checkKeyLength($key); + $responseString = $memcache->delete($key); + break; + default: + case Request::HEAD: + throw new InvalidException('Method '.$request->getMethod().' is not supported in memcache transport'); + } + + $end = microtime(true); + + $response = new Response($responseString); + $response->setQueryTime($end - $start); + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + return $response; + } + + /** + * Check if key that will be used dont exceed 250 symbols. + * + * @param string $key + * + * @throws Elastica\Exception\Connection\MemcacheException If key is too long + */ + private function _checkKeyLength($key) + { + if (strlen($key) >= self::MAX_KEY_LENGTH) { + throw new MemcacheException('Memcache key is too long'); + } + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Null.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Null.php new file mode 100644 index 00000000..70dd9af1 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Null.php @@ -0,0 +1,13 @@ +<?php +namespace Elastica\Transport; + +/** + * Elastica Null Transport object. + * + * This class is for backward compatibility reason for all php < 7 versions. For PHP 7 and above use NullTransport as Null is reserved. + * + * @author James Boehmer <james.boehmer@jamesboehmer.com> + */ +class Null extends NullTransport +{ +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/NullTransport.php b/vendor/ruflin/elastica/lib/Elastica/Transport/NullTransport.php new file mode 100644 index 00000000..d24f2110 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/NullTransport.php @@ -0,0 +1,46 @@ +<?php +namespace Elastica\Transport; + +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; + +/** + * Elastica Null Transport object. + * + * This is used in case you just need a test transport that doesn't do any connection to an elasticsearch + * host but still returns a valid response object + * + * @author James Boehmer <james.boehmer@jamesboehmer.com> + */ +class NullTransport extends AbstractTransport +{ + /** + * Null transport. + * + * @param \Elastica\Request $request + * @param array $params Hostname, port, path, ... + * + * @return \Elastica\Response Response empty object + */ + public function exec(Request $request, array $params) + { + $response = array( + 'took' => 0, + 'timed_out' => false, + '_shards' => array( + 'total' => 0, + 'successful' => 0, + 'failed' => 0, + ), + 'hits' => array( + 'total' => 0, + 'max_score' => null, + 'hits' => array(), + ), + 'params' => $params, + ); + + return new Response(JSON::stringify($response)); + } +} diff --git a/vendor/ruflin/elastica/lib/Elastica/Transport/Thrift.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Thrift.php new file mode 100644 index 00000000..5790f665 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Thrift.php @@ -0,0 +1,176 @@ +<?php +namespace Elastica\Transport; + +use Elastica\Connection; +use Elastica\Exception\Connection\ThriftException; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\Exception\RuntimeException; +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; +use Elasticsearch\Method; +use Elasticsearch\RestClient; +use Elasticsearch\RestRequest; +use Elasticsearch\RestResponse; +use Thrift\Exception\TException; +use Thrift\Protocol\TBinaryProtocolAccelerated; +use Thrift\Transport\TBufferedTransport; +use Thrift\Transport\TFramedTransport; +use Thrift\Transport\TSocket; + +/** + * Elastica Thrift Transport object. + * + * @author Mikhail Shamin <munk13@gmail.com> + * + * @deprecated The thrift transport is deprecated as of ES 1.5, and will be removed in ES 2.0 + */ +class Thrift extends AbstractTransport +{ + /** + * @var RestClient[] + */ + protected $_clients = array(); + + /** + * Construct transport. + * + * @param \Elastica\Connection $connection Connection object + * + * @throws \Elastica\Exception\RuntimeException + */ + public function __construct(Connection $connection = null) + { + parent::__construct($connection); + if (!class_exists('Elasticsearch\\RestClient')) { + throw new RuntimeException('Elasticsearch\\RestClient class not found. Check that suggested package munkie/elasticsearch-thrift-php is required in composer.json'); + } + } + + /** + * @param string $host + * @param int $port + * @param int $sendTimeout msec + * @param int $recvTimeout msec + * @param bool $framedTransport + * + * @return \Elasticsearch\RestClient + */ + protected function _createClient($host, $port, $sendTimeout = null, $recvTimeout = null, $framedTransport = false) + { + $socket = new TSocket($host, $port, true); + + if (null !== $sendTimeout) { + $socket->setSendTimeout($sendTimeout); + } + + if (null !== $recvTimeout) { + $socket->setRecvTimeout($recvTimeout); + } + + if ($framedTransport) { + $transport = new TFramedTransport($socket); + } else { + $transport = new TBufferedTransport($socket); + } + $protocol = new TBinaryProtocolAccelerated($transport); + + $client = new RestClient($protocol); + + $transport->open(); + + return $client; + } + + /** + * @param string $host + * @param int $port + * @param int $sendTimeout + * @param int $recvTimeout + * @param bool $framedTransport + * + * @return \Elasticsearch\RestClient + */ + protected function _getClient($host, $port, $sendTimeout = null, $recvTimeout = null, $framedTransport = false) + { + $key = $host.':'.$port; + if (!isset($this->_clients[$key])) { + $this->_clients[$key] = $this->_createClient($host, $port, $sendTimeout, $recvTimeout, $framedTransport); + } + + return $this->_clients[$key]; + } + + /** + * Makes calls to the elasticsearch server. + * + * @param \Elastica\Request $request + * @param array $params Host, Port, ... + * + * @throws \Elastica\Exception\Connection\ThriftException + * @throws \Elastica\Exception\ResponseException + * + * @return \Elastica\Response Response object + */ + public function exec(Request $request, array $params) + { + $connection = $this->getConnection(); + + $sendTimeout = $connection->hasConfig('sendTimeout') ? $connection->getConfig('sendTimeout') : null; + $recvTimeout = $connection->hasConfig('recvTimeout') ? $connection->getConfig('recvTimeout') : null; + $framedTransport = $connection->hasConfig('framedTransport') ? (bool) $connection->getConfig('framedTransport') : false; + + try { + $client = $this->_getClient( + $connection->getHost(), + $connection->getPort(), + $sendTimeout, + $recvTimeout, + $framedTransport + ); + + $restRequest = new RestRequest(); + $restRequest->method = array_search($request->getMethod(), Method::$__names); + $restRequest->uri = $request->getPath(); + + $query = $request->getQuery(); + if (!empty($query)) { + $restRequest->parameters = $query; + } + + $data = $request->getData(); + if (!empty($data) || '0' === $data) { + if (is_array($data)) { + $content = JSON::stringify($data); + } else { + $content = $data; + } + $restRequest->body = $content; + } + + /* @var $result RestResponse */ + $start = microtime(true); + + $result = $client->execute($restRequest); + $response = new Response($result->body); + + $end = microtime(true); + } catch (TException $e) { + $response = new Response(''); + throw new ThriftException($e, $request, $response); + } + + $response->setQueryTime($end - $start); + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + return $response; + } +} |