/**
  * {@inheritdoc}
  */
 public function toArray()
 {
     try {
         return JSON::parse($this->__toString());
     } catch (JSONParseException $e) {
         throw new InvalidException('The query produced is invalid');
     }
 }

 /**
  * {@inheritdoc}
  */
 public function toArray()
 {
     try {
         return JSON::parse($input = $this->__toString());
     } catch (JSONParseException $e) {
         throw new InvalidException(sprintf('The produced query is not a valid json string : "%s"', $input));
     }
 }

 /**
  * Log a message.
  *
  * @param mixed  $level
  * @param string $message
  * @param array  $context
  *
  * @return null|void
  */
 public function log($level, $message, array $context = array())
 {
     $context['error_message'] = $message;
     $this->_lastMessage = JSON::stringify($context);
     if (!empty($this->_log) && is_string($this->_log)) {
         error_log($this->_lastMessage . PHP_EOL, 3, $this->_log);
     } else {
         error_log($this->_lastMessage);
     }
 }

 /**
  * 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);
     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);
     if ($connection->hasConfig('bigintConversion')) {
         $response->setJsonBigintConversion($connection->getConfig('bigintConversion'));
     }
     $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;
 }

 /**
  * 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);
     if (\Elastica\Util::debugEnabled()) {
         $response->setQueryTime($end - $start);
     }
     if ($response->hasError()) {
         throw new ResponseException($request, $response);
     }
     if ($response->hasFailedShards()) {
         throw new PartialShardFailureException($request, $response);
     }
     return $response;
 }

 /**
  * 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 (isset($data) && !empty($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()));
     }
 }

 /**
  * 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)) {
         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;
 }

 /**
  * @group functional
  */
 public function testPartialFailure()
 {
     $this->_checkScriptInlineSetting();
     $client = $this->_getClient();
     $index = $client->getIndex('elastica_partial_failure');
     $index->create(array('index' => array('number_of_shards' => 5, 'number_of_replicas' => 0)), true);
     $type = $index->getType('folks');
     $type->addDocument(new Document('', array('name' => 'ruflin')));
     $type->addDocument(new Document('', array('name' => 'bobrik')));
     $type->addDocument(new Document('', array('name' => 'kimchy')));
     $index->refresh();
     $query = Query::create(array('query' => array('filtered' => array('filter' => array('script' => array('script' => 'doc["undefined"] > 8'))))));
     try {
         $index->search($query);
         $this->fail('PartialShardFailureException should have been thrown');
     } catch (PartialShardFailureException $e) {
         $resultSet = new ResultSet($e->getResponse(), $query);
         $this->assertEquals(0, count($resultSet->getResults()));
         $message = JSON::parse($e->getMessage());
         $this->assertTrue(isset($message['failures']), 'Failures are absent');
         $this->assertGreaterThan(0, count($message['failures']), 'Failures are empty');
     }
 }

 /**
  * 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);
         if ($connection->hasCompression()) {
             // An "Accept-Encoding" header containing all supported encoding types is sent
             // Curl will decode the response automatically
             curl_setopt($conn, CURLOPT_ENCODING, '');
             // Let's precise that the request is also compressed
             curl_setopt($conn, CURLOPT_HTTPHEADER, array('Content-Encoding: gzip'));
             // Let's compress the request body,
             curl_setopt($conn, CURLOPT_POSTFIELDS, gzencode($content));
         } else {
             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 ($connection->hasConfig('bigintConversion')) {
         $response->setJsonBigintConversion($connection->getConfig('bigintConversion'));
     }
     if ($response->hasError()) {
//.........这里部分代码省略.........

 /**
  * @param  \Elastica\Search $search
  * @return string
  */
 protected function _getSearchData(BaseSearch $search)
 {
     $header = $this->_getSearchDataHeader($search);
     $header = empty($header) ? new \stdClass() : $header;
     $query = $search->getQuery();
     $data = JSON::stringify($header) . "\n";
     $data .= JSON::stringify($query->toArray()) . "\n";
     return $data;
 }

 /**
  * Converts Request to Curl console command
  *
  * @param Request $request
  * @return string
  */
 public static function convertRequestToCurlCommand(Request $request)
 {
     $message = 'curl -X' . strtoupper($request->getMethod()) . ' ';
     $message .= '\'http://' . $request->getConnection()->getHost() . ':' . $request->getConnection()->getPort() . '/';
     $message .= $request->getPath();
     $query = $request->getQuery();
     if (!empty($query)) {
         $message .= '?' . http_build_query($query);
     }
     $message .= '\'';
     $data = $request->getData();
     if (!empty($data)) {
         $message .= ' -d \'' . JSON::stringify($data) . '\'';
     }
     return $message;
 }

 /**
  * 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));
 }

 /**
  * @return string
  */
 public function toString()
 {
     $string = JSON::stringify($this->getActionMetadata(), JSON_FORCE_OBJECT) . Bulk::DELIMITER;
     if ($this->hasSource()) {
         $source = $this->getSource();
         if (is_string($source)) {
             $string .= $source;
         } elseif (is_array($source) && array_key_exists('doc', $source) && is_string($source['doc'])) {
             $docAsUpsert = isset($source['doc_as_upsert']) ? ', "doc_as_upsert": ' . $source['doc_as_upsert'] : '';
             $string .= '{"doc": ' . $source['doc'] . $docAsUpsert . '}';
         } else {
             $string .= JSON::stringify($source, 'JSON_ELASTICSEARCH');
         }
         $string .= Bulk::DELIMITER;
     }
     return $string;
 }

 /**
  * Construct Exception.
  *
  * @param \Elastica\Request  $request
  * @param \Elastica\Response $response
  */
 public function __construct(Request $request, Response $response)
 {
     parent::__construct($request, $response);
     $shardsStatistics = $response->getShardsStatistics();
     $this->message = JSON::stringify($shardsStatistics);
 }

 public function write(array $document)
 {
     $indexOp = array('index' => array('_type' => $document['_type'], '_id' => $document['_id']));
     // We use Elastica wrapper around json_encode.
     // Depending on PHP version JSON_ESCAPE_UNICODE will be used
     $this->writeLine(JSON::stringify($indexOp));
     $this->writeLine(JSON::stringify($document['_source']));
 }

 /**
  * Response data array.
  *
  * @return array Response data array
  */
 public function getData()
 {
     if ($this->_response == null) {
         $response = $this->_responseString;
         if ($response === false) {
             $this->_error = true;
         } else {
             try {
                 if ($this->getJsonBigintConversion()) {
                     $response = JSON::parse($response, true, 512, JSON_BIGINT_AS_STRING);
                 } else {
                     $response = JSON::parse($response);
                 }
             } catch (JSONParseException $e) {
                 // leave response as is if parse fails
             }
         }
         if (empty($response)) {
             $response = [];
         }
         if (is_string($response)) {
             $response = ['message' => $response];
         }
         $this->_response = $response;
     }
     return $this->_response;
 }

 /**
  * Response data array.
  *
  * @return array Response data array
  */
 public function getData()
 {
     if ($this->_response == null) {
         $response = $this->_responseString;
         if ($response === false) {
             $this->_error = true;
         } else {
             try {
                 $response = JSON::parse($response);
             } catch (JSONParseException $e) {
                 // leave reponse as is if parse fails
             }
         }
         if (empty($response)) {
             $response = array();
         }
         if (is_string($response)) {
             $response = array('message' => $response);
         }
         $this->_response = $response;
     }
     return $this->_response;
 }

 /**
  * @param \Elastica\Search $search
  *
  * @return string
  */
 protected function _getSearchData(BaseSearch $search)
 {
     $header = $this->_getSearchDataHeader($search);
     $header = empty($header) ? new \stdClass() : $header;
     $query = $search->getQuery();
     // Keep other query options as part of the search body
     $queryOptions = array_diff_key($search->getOptions(), array_flip(self::$HEADER_OPTIONS));
     $data = JSON::stringify($header) . "\n";
     $data .= JSON::stringify($query->toArray() + $queryOptions) . "\n";
     return $data;
 }

 /**
  * 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;
 }

 /**
  * 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));
 }