diff options
Diffstat (limited to 'vendor/ruflin/elastica/lib/Elastica/Transport')
7 files changed, 805 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..d2ce0fb2 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/AbstractTransport.php @@ -0,0 +1,109 @@ +<?php + +namespace Elastica\Transport; + +use Elastica\Connection; +use Elastica\Request; +use Elastica\Exception\InvalidException; +use Elastica\Param; + +/** + * Elastica Abstract Transport object + * + * @category Xodoa + * @package Elastica + * @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 + */ + public function setConnection(Connection $connection) + { + $this->_connection = $connection; + } + + /** + * 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 AbstractTransport) { + $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..96ad8971 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Guzzle.php @@ -0,0 +1,179 @@ +<?php + +namespace Elastica\Transport; + +use Elastica\Exception\Connection\HttpException; +use Elastica\Exception\Connection\GuzzleException; +use Elastica\Exception\PartialShardFailureException; +use Elastica\Exception\ResponseException; +use Elastica\Exception\InvalidException; +use Elastica\Connection; +use Elastica\Request; +use Elastica\Response; +use Elastica\JSON; +use GuzzleHttp\Client; +use GuzzleHttp\Exception\TransferException; +use GuzzleHttp\Exception\ClientException; +use GuzzleHttp\Stream\Stream; + +/** + * Elastica Guzzle Transport object + * + * @package Elastica + * @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(); + + try { + $client = $this->_getGuzzleClient($this->_getBaseUrl($connection), $connection->isPersistent()); + + $options = array(); + if ($connection->getTimeout()) { + $options['timeout'] = $connection->getTimeout(); + } + + if ($connection->getProxy()) { + $options['proxy'] = $connection->getProxy(); + } + + $req = $client->createRequest($request->getMethod(), $this->_getActionPath($request), $options); + $req->setHeaders($connection->hasConfig('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)); + } + + $start = microtime(true); + $res = $client->send($req); + $end = microtime(true); + + $response = new Response((string)$res->getBody(), $res->getStatusCode()); + + if (defined('DEBUG') && DEBUG) { + $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; + + } catch (ClientException $e) { + // ignore 4xx errors + } catch (TransferException $e) { + throw new GuzzleException($e, $request, new Response($e->getMessage())); + } + + } + + /** + * 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..5606dbbc --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Http.php @@ -0,0 +1,188 @@ +<?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 + * + * @category Xodoa + * @package Elastica + * @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); + + $proxy = $connection->getProxy(); + 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); + + if (defined('DEBUG') && DEBUG) { + // Track request headers when in debug mode + curl_setopt($conn, CURLINFO_HEADER_OUT, true); + } + + $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($this->_getConnection(), CURLINFO_HTTP_CODE)); + + if (defined('DEBUG') && DEBUG) { + $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/Https.php b/vendor/ruflin/elastica/lib/Elastica/Transport/Https.php new file mode 100644 index 00000000..64704039 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Https.php @@ -0,0 +1,30 @@ +<?php + +namespace Elastica\Transport; + +/** + * Elastica Http Transport object + * + * @category Xodoa + * @package Elastica + * @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..cf047b58 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Memcache.php @@ -0,0 +1,82 @@ +<?php + +namespace Elastica\Transport; + +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 + * + * @category Xodoa + * @package Elastica + * @author Nicolas Ruflin <spam@ruflin.com> + */ +class Memcache extends AbstractTransport +{ + /** + * 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()); + + // Finds right function name + $function = strtolower($request->getMethod()); + + $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 = ''; + + switch ($function) { + case 'post': + case 'put': + $memcache->set($request->getPath(), $content); + break; + case 'get': + $responseString = $memcache->get($request->getPath() . '?source=' . $content); + break; + case 'delete': + break; + default: + throw new InvalidException('Method ' . $function . ' is not supported in memcache transport'); + + } + + $response = new Response($responseString); + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + return $response; + } +} 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..704e0e12 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Null.php @@ -0,0 +1,44 @@ +<?php + +namespace Elastica\Transport; + +use Elastica\JSON; +use Elastica\Request; +use Elastica\Response; + +/** + * Elastica Null Transport object + * + * @package Elastica + * @author James Boehmer <james.boehmer@jamesboehmer.com> + */ +class Null 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..f58c51a2 --- /dev/null +++ b/vendor/ruflin/elastica/lib/Elastica/Transport/Thrift.php @@ -0,0 +1,173 @@ +<?php + +namespace Elastica\Transport; + +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 Elastica\Connection; +use Elasticsearch\Method; +use Elasticsearch\RestResponse; +use Elasticsearch\RestClient; +use Elasticsearch\RestRequest; +use Thrift\Transport\TSocket; +use Thrift\Transport\TFramedTransport; +use Thrift\Transport\TBufferedTransport; +use Thrift\Protocol\TBinaryProtocolAccelerated; +use Thrift\Exception\TException; + +/** + * Elastica Thrift Transport object + * + * @category Xodoa + * @package Elastica + * @author Mikhail Shamin <munk13@gmail.com> + */ +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); + } + + if (defined('DEBUG') && DEBUG) { + $response->setQueryTime($end - $start); + } + + if ($response->hasError()) { + throw new ResponseException($request, $response); + } + + if ($response->hasFailedShards()) { + throw new PartialShardFailureException($request, $response); + } + + return $response; + } +} |