PhpiredisConnection.php 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  1. <?php
  2. /*
  3. * This file is part of the Predis package.
  4. *
  5. * (c) Daniele Alessandri <suppakilla@gmail.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Predis\Connection;
  11. use Predis\Command\CommandInterface;
  12. use Predis\ConnectionParametersInterface;
  13. use Predis\ResponseError;
  14. use Predis\ResponseQueued;
  15. use Predis\ClientException;
  16. use Predis\ServerException;
  17. use Predis\NotSupportedException;
  18. /**
  19. * This class provides the implementation of a Predis connection that uses the
  20. * PHP socket extension for network communication and wraps the phpiredis C
  21. * extension (PHP bindings for hiredis) to parse the Redis protocol. Everything
  22. * is highly experimental (even the very same phpiredis since it is quite new),
  23. * so use it at your own risk.
  24. *
  25. * This class is mainly intended to provide an optional low-overhead alternative
  26. * for processing replies from Redis compared to the standard pure-PHP classes.
  27. * Differences in speed when dealing with short inline replies are practically
  28. * nonexistent, the actual speed boost is for long multibulk replies when this
  29. * protocol processor can parse and return replies very fast.
  30. *
  31. * For instructions on how to build and install the phpiredis extension, please
  32. * consult the repository of the project.
  33. *
  34. * The connection parameters supported by this class are:
  35. *
  36. * - scheme: it can be either 'tcp' or 'unix'.
  37. * - host: hostname or IP address of the server.
  38. * - port: TCP port of the server.
  39. * - timeout: timeout to perform the connection.
  40. * - read_write_timeout: timeout of read / write operations.
  41. * - throw_errors: -ERR replies treated as exceptions.
  42. *
  43. * @link http://github.com/seppo0010/phpiredis
  44. * @author Daniele Alessandri <suppakilla@gmail.com>
  45. */
  46. class PhpiredisConnection extends AbstractConnection
  47. {
  48. const ERR_MSG_EXTENSION = 'The %s extension must be loaded in order to be able to use this connection class';
  49. private $reader;
  50. /**
  51. * {@inheritdoc}
  52. */
  53. public function __construct(ConnectionParametersInterface $parameters)
  54. {
  55. $this->checkExtensions();
  56. parent::__construct($parameters);
  57. }
  58. /**
  59. * Disconnects from the server and destroys the underlying resource and the
  60. * protocol reader resource when PHP's garbage collector kicks in.
  61. */
  62. public function __destruct()
  63. {
  64. phpiredis_reader_destroy($this->reader);
  65. parent::__destruct();
  66. }
  67. /**
  68. * Checks if the cURL and phpiredis extensions are loaded in PHP.
  69. */
  70. private function checkExtensions()
  71. {
  72. if (!function_exists('socket_create')) {
  73. throw new NotSupportedException(sprintf(self::ERR_MSG_EXTENSION, 'socket'));
  74. }
  75. if (!function_exists('phpiredis_reader_create')) {
  76. throw new NotSupportedException(sprintf(self::ERR_MSG_EXTENSION, 'phpiredis'));
  77. }
  78. }
  79. /**
  80. * {@inheritdoc}
  81. */
  82. protected function checkParameters(ConnectionParametersInterface $parameters)
  83. {
  84. if ($parameters->iterable_multibulk === true) {
  85. $this->onInvalidOption('iterable_multibulk', $parameters);
  86. }
  87. if ($parameters->persistent === true) {
  88. $this->onInvalidOption('persistent', $parameters);
  89. }
  90. return parent::checkParameters($parameters);
  91. }
  92. /**
  93. * Initializes the protocol reader resource.
  94. *
  95. * @param Boolean $throw_errors Specify if Redis errors throw exceptions.
  96. */
  97. private function initializeReader($throw_errors = true) {
  98. $reader = phpiredis_reader_create();
  99. phpiredis_reader_set_status_handler($reader, $this->getStatusHandler());
  100. phpiredis_reader_set_error_handler($reader, $this->getErrorHandler($throw_errors));
  101. $this->reader = $reader;
  102. }
  103. /**
  104. * {@inheritdoc}
  105. */
  106. protected function initializeProtocol(ConnectionParametersInterface $parameters)
  107. {
  108. $this->initializeReader($parameters->throw_errors);
  109. }
  110. /**
  111. * Gets the handler used by the protocol reader to handle status replies.
  112. *
  113. * @return \Closure
  114. */
  115. private function getStatusHandler()
  116. {
  117. return function($payload) {
  118. switch ($payload) {
  119. case 'OK':
  120. return true;
  121. case 'QUEUED':
  122. return new ResponseQueued();
  123. default:
  124. return $payload;
  125. }
  126. };
  127. }
  128. /**
  129. * Gets the handler used by the protocol reader to handle Redis errors.
  130. *
  131. * @param Boolean $throw_errors Specify if Redis errors throw exceptions.
  132. * @return \Closure
  133. */
  134. private function getErrorHandler($throwErrors = true)
  135. {
  136. if ($throwErrors) {
  137. return function($errorMessage) {
  138. throw new ServerException($errorMessage);
  139. };
  140. }
  141. return function($errorMessage) {
  142. return new ResponseError($errorMessage);
  143. };
  144. }
  145. /**
  146. * Helper method used to throw exceptions on socket errors.
  147. */
  148. private function emitSocketError()
  149. {
  150. $errno = socket_last_error();
  151. $errstr = socket_strerror($errno);
  152. $this->disconnect();
  153. $this->onConnectionError(trim($errstr), $errno);
  154. }
  155. /**
  156. * {@inheritdoc}
  157. */
  158. protected function createResource()
  159. {
  160. $parameters = $this->parameters;
  161. $isUnix = $this->parameters->scheme === 'unix';
  162. $domain = $isUnix ? AF_UNIX : AF_INET;
  163. $protocol = $isUnix ? 0 : SOL_TCP;
  164. $socket = @call_user_func('socket_create', $domain, SOCK_STREAM, $protocol);
  165. if (!is_resource($socket)) {
  166. $this->emitSocketError();
  167. }
  168. $this->setSocketOptions($socket, $parameters);
  169. return $socket;
  170. }
  171. /**
  172. * Sets options on the socket resource from the connection parameters.
  173. *
  174. * @param resource $socket Socket resource.
  175. * @param ConnectionParametersInterface $parameters Parameters used to initialize the connection.
  176. */
  177. private function setSocketOptions($socket, ConnectionParametersInterface $parameters)
  178. {
  179. if ($parameters->scheme !== 'tcp') {
  180. return;
  181. }
  182. if (!socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1)) {
  183. $this->emitSocketError();
  184. }
  185. if (!socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1)) {
  186. $this->emitSocketError();
  187. }
  188. if (isset($parameters->read_write_timeout)) {
  189. $rwtimeout = $parameters->read_write_timeout;
  190. $timeoutSec = floor($rwtimeout);
  191. $timeoutUsec = ($rwtimeout - $timeoutSec) * 1000000;
  192. $timeout = array(
  193. 'sec' => $timeoutSec,
  194. 'usec' => $timeoutUsec,
  195. );
  196. if (!socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout)) {
  197. $this->emitSocketError();
  198. }
  199. if (!socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, $timeout)) {
  200. $this->emitSocketError();
  201. }
  202. }
  203. }
  204. /**
  205. * Gets the address from the connection parameters.
  206. *
  207. * @param ConnectionParametersInterface $parameters Parameters used to initialize the connection.
  208. * @return string
  209. */
  210. private function getAddress(ConnectionParametersInterface $parameters)
  211. {
  212. if ($parameters->scheme === 'unix') {
  213. return $parameters->path;
  214. }
  215. $host = $parameters->host;
  216. if (ip2long($host) === false) {
  217. if (($address = gethostbyname($host)) === $host) {
  218. $this->onConnectionError("Cannot resolve the address of $host");
  219. }
  220. return $address;
  221. }
  222. return $host;
  223. }
  224. /**
  225. * Opens the actual connection to the server with a timeout.
  226. *
  227. * @param ConnectionParametersInterface $parameters Parameters used to initialize the connection.
  228. * @return string
  229. */
  230. private function connectWithTimeout(ConnectionParametersInterface $parameters) {
  231. $host = self::getAddress($parameters);
  232. $socket = $this->getResource();
  233. socket_set_nonblock($socket);
  234. if (@socket_connect($socket, $host, $parameters->port) === false) {
  235. $error = socket_last_error();
  236. if ($error != SOCKET_EINPROGRESS && $error != SOCKET_EALREADY) {
  237. $this->emitSocketError();
  238. }
  239. }
  240. socket_set_block($socket);
  241. $null = null;
  242. $selectable = array($socket);
  243. $timeout = $parameters->timeout;
  244. $timeoutSecs = floor($timeout);
  245. $timeoutUSecs = ($timeout - $timeoutSecs) * 1000000;
  246. $selected = socket_select($selectable, $selectable, $null, $timeoutSecs, $timeoutUSecs);
  247. if ($selected === 2) {
  248. $this->onConnectionError('Connection refused', SOCKET_ECONNREFUSED);
  249. }
  250. if ($selected === 0) {
  251. $this->onConnectionError('Connection timed out', SOCKET_ETIMEDOUT);
  252. }
  253. if ($selected === false) {
  254. $this->emitSocketError();
  255. }
  256. }
  257. /**
  258. * {@inheritdoc}
  259. */
  260. public function connect()
  261. {
  262. parent::connect();
  263. $this->connectWithTimeout($this->parameters);
  264. if (count($this->initCmds) > 0) {
  265. $this->sendInitializationCommands();
  266. }
  267. }
  268. /**
  269. * {@inheritdoc}
  270. */
  271. public function disconnect()
  272. {
  273. if ($this->isConnected()) {
  274. socket_close($this->getResource());
  275. parent::disconnect();
  276. }
  277. }
  278. /**
  279. * Sends the initialization commands to Redis when the connection is opened.
  280. */
  281. private function sendInitializationCommands()
  282. {
  283. foreach ($this->initCmds as $command) {
  284. $this->writeCommand($command);
  285. }
  286. foreach ($this->initCmds as $command) {
  287. $this->readResponse($command);
  288. }
  289. }
  290. /**
  291. * {@inheritdoc}
  292. */
  293. private function write($buffer)
  294. {
  295. $socket = $this->getResource();
  296. while (($length = strlen($buffer)) > 0) {
  297. $written = socket_write($socket, $buffer, $length);
  298. if ($length === $written) {
  299. return;
  300. }
  301. if ($written === false) {
  302. $this->onConnectionError('Error while writing bytes to the server');
  303. }
  304. $buffer = substr($buffer, $written);
  305. }
  306. }
  307. /**
  308. * {@inheritdoc}
  309. */
  310. public function read()
  311. {
  312. $socket = $this->getResource();
  313. $reader = $this->reader;
  314. while (($state = phpiredis_reader_get_state($reader)) === PHPIREDIS_READER_STATE_INCOMPLETE) {
  315. if (@socket_recv($socket, $buffer, 4096, 0) === false || $buffer === '') {
  316. $this->emitSocketError();
  317. }
  318. phpiredis_reader_feed($reader, $buffer);
  319. }
  320. if ($state === PHPIREDIS_READER_STATE_COMPLETE) {
  321. return phpiredis_reader_get_reply($reader);
  322. }
  323. else {
  324. $this->onProtocolError(phpiredis_reader_get_error($reader));
  325. }
  326. }
  327. /**
  328. * {@inheritdoc}
  329. */
  330. public function writeCommand(CommandInterface $command)
  331. {
  332. $cmdargs = $command->getArguments();
  333. array_unshift($cmdargs, $command->getId());
  334. $this->write(phpiredis_format_command($cmdargs));
  335. }
  336. /**
  337. * {@inheritdoc}
  338. */
  339. public function __wakeup()
  340. {
  341. $this->initializeProtocol($this->getParameters());
  342. }
  343. }