diff --git a/lib/Predis/Network/WebdisConnection.php b/lib/Predis/Network/WebdisConnection.php new file mode 100644 index 00000000..79516da6 --- /dev/null +++ b/lib/Predis/Network/WebdisConnection.php @@ -0,0 +1,143 @@ +checkExtensions(); + if ($parameters->scheme !== 'http') { + throw new \InvalidArgumentException("Invalid scheme: {$parameters->scheme}"); + } + $this->_parameters = $parameters; + $this->_webdisUrl = "{$parameters->scheme}://{$parameters->host}:{$parameters->port}"; + $this->_reader = $this->initializeReader($parameters); + } + + public function __destruct() { + phpiredis_reader_destroy($this->_reader); + } + + private function checkExtensions() { + if (!class_exists("HttpRequest")) { + throw new ClientException(sprintf(ERR_MSG_EXTENSION, 'http')); + } + if (!function_exists('phpiredis_reader_create')) { + throw new ClientException(sprintf(ERR_MSG_EXTENSION, 'phpiredis')); + } + } + + private function initializeReader(IConnectionParameters $parameters) { + $throwErrors = $parameters->throw_errors; + $reader = phpiredis_reader_create(); + phpiredis_reader_set_status_handler($reader, $this->getStatusHandler()); + phpiredis_reader_set_error_handler($reader, $this->getErrorHandler($throwErrors)); + return $reader; + } + + private function getStatusHandler() { + return function($payload) { + return $payload === 'OK' ? true : $payload; + }; + } + + private function getErrorHandler($throwErrors) { + if ($throwErrors) { + return function($errorMessage) { + throw new ServerException($errorMessage); + }; + } + return function($errorMessage) { + return new ResponseError($errorMessage); + }; + } + + private static function argumentsSerializer($str, $arg) { + $str .= '/' . urlencode($arg); + return $str; + } + + public function connect() { + // NOOP + } + + public function disconnect() { + // NOOP + } + + public function isConnected() { + return true; + } + + public function writeCommand(ICommand $command) { + self::throwNotImplementedException(__CLASS__, __FUNCTION__); + } + + public function readResponse(ICommand $command) { + self::throwNotImplementedException(__CLASS__, __FUNCTION__); + } + + public function executeCommand(ICommand $command) { + $params = $this->_parameters; + $arguments = array_reduce($command->getArguments(), 'self::argumentsSerializer'); + + $request = new HttpRequest($this->_webdisUrl, HttpRequest::METH_POST); + $request->setBody(sprintf('%s%s.raw', $command->getId(), $arguments)); + $request->send(); + + phpiredis_reader_feed($this->_reader, $request->getResponseBody()); + $reply = phpiredis_reader_get_reply($this->_reader); + return isset($reply->skipParse) ? $reply : $command->parseResponse($reply); + } + + public function getResource() { + self::throwNotImplementedException(__CLASS__, __FUNCTION__); + } + + public function getParameters() { + return $this->_parameters; + } + + public function pushInitCommand(ICommand $command) { + self::throwNotImplementedException(__CLASS__, __FUNCTION__); + } + + public function read() { + self::throwNotImplementedException(__CLASS__, __FUNCTION__); + } + + public function __toString() { + return "{$this->_parameters->host}:{$this->_parameters->port}"; + } +}