Predis.php 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122
  1. <?php
  2. namespace Predis;
  3. class PredisException extends \Exception { }
  4. class ClientException extends PredisException { }
  5. class ServerException extends PredisException { }
  6. class MalformedServerResponse extends ServerException { }
  7. /* ------------------------------------------------------------------------- */
  8. class Client {
  9. // TODO: command arguments should be sanitized or checked for bad arguments
  10. // (e.g. CRLF in keys for inline commands)
  11. private $_connection, $_registeredCommands;
  12. public function __construct($host = Connection::DEFAULT_HOST, $port = Connection::DEFAULT_PORT) {
  13. $this->_registeredCommands = self::initializeDefaultCommands();
  14. $this->_connection = new Connection($host, $port);
  15. }
  16. public function __destruct() {
  17. $this->_connection->disconnect();
  18. }
  19. public static function createCluster(/* arguments */) {
  20. $cluster = new ConnectionCluster();
  21. foreach (func_get_args() as $parameters) {
  22. $cluster->add(new Connection($parameters['host'], $parameters['port']));
  23. }
  24. $client = new Client();
  25. $client->setConnection($cluster);
  26. return $client;
  27. }
  28. private function setConnection(IConnection $connection) {
  29. $this->_connection = $connection;
  30. }
  31. public function connect() {
  32. $this->_connection->connect();
  33. }
  34. public function disconnect() {
  35. $this->_connection->disconnect();
  36. }
  37. public function isConnected() {
  38. return $this->_connection->isConnected();
  39. }
  40. public function getConnection() {
  41. return $this->_connection;
  42. }
  43. public function __call($method, $arguments) {
  44. $command = $this->createCommandInstance($method, $arguments);
  45. return $this->executeCommand($command);
  46. }
  47. public function createCommandInstance($method, $arguments) {
  48. $commandClass = $this->_registeredCommands[$method];
  49. if ($commandClass === null) {
  50. throw new ClientException("'$method' is not a registered Redis command");
  51. }
  52. $command = new $commandClass();
  53. $command->setArgumentsArray($arguments);
  54. return $command;
  55. }
  56. public function executeCommand(Command $command) {
  57. $this->_connection->writeCommand($command);
  58. if ($command->closesConnection()) {
  59. return $this->_connection->disconnect();
  60. }
  61. return $this->_connection->readResponse($command);
  62. }
  63. public function rawCommand($rawCommandData, $closesConnection = false) {
  64. // TODO: rather than check the type of a connection instance, we should
  65. // check if it does respond to the rawCommand method.
  66. if (is_a($this->_connection, '\Predis\ConnectionCluster')) {
  67. throw new ClientException('Cannot send raw commands when connected to a cluster of Redis servers');
  68. }
  69. return $this->_connection->rawCommand($rawCommandData, $closesConnection);
  70. }
  71. public function pipeline(\Closure $pipelineBlock = null) {
  72. $pipeline = new CommandPipeline($this);
  73. return $pipelineBlock !== null ? $pipeline->execute($pipelineBlock) : $pipeline;
  74. }
  75. public function registerCommands(Array $commands) {
  76. foreach ($commands as $command => $aliases) {
  77. $this->registerCommand($command, $aliases);
  78. }
  79. }
  80. public function registerCommand($command, $aliases) {
  81. $commandReflection = new \ReflectionClass($command);
  82. if (!$commandReflection->isSubclassOf('\Predis\Command')) {
  83. throw new ClientException("Cannot register '$command' as it is not a valid Redis command");
  84. }
  85. if (is_array($aliases)) {
  86. foreach ($aliases as $alias) {
  87. $this->_registeredCommands[$alias] = $command;
  88. }
  89. }
  90. else {
  91. $this->_registeredCommands[$aliases] = $command;
  92. }
  93. }
  94. private static function initializeDefaultCommands() {
  95. // NOTE: we don't use \Predis\Client::registerCommands for performance reasons.
  96. return array(
  97. /* miscellaneous commands */
  98. 'ping' => '\Predis\Commands\Ping',
  99. 'echo' => '\Predis\Commands\DoEcho',
  100. 'auth' => '\Predis\Commands\Auth',
  101. /* connection handling */
  102. 'quit' => '\Predis\Commands\Quit',
  103. /* commands operating on string values */
  104. 'set' => '\Predis\Commands\Set',
  105. 'setnx' => '\Predis\Commands\SetPreserve',
  106. 'setPreserve' => '\Predis\Commands\SetPreserve',
  107. 'mset' => '\Predis\Commands\SetMultiple',
  108. 'setMultiple' => '\Predis\Commands\SetMultiple',
  109. 'msetnx' => '\Predis\Commands\SetMultiplePreserve',
  110. 'setMultiplePreserve' => '\Predis\Commands\SetMultiplePreserve',
  111. 'get' => '\Predis\Commands\Get',
  112. 'mget' => '\Predis\Commands\GetMultiple',
  113. 'getMultiple' => '\Predis\Commands\GetMultiple',
  114. 'getset' => '\Predis\Commands\GetSet',
  115. 'getSet' => '\Predis\Commands\GetSet',
  116. 'incr' => '\Predis\Commands\Increment',
  117. 'increment' => '\Predis\Commands\Increment',
  118. 'incrby' => '\Predis\Commands\IncrementBy',
  119. 'incrementBy' => '\Predis\Commands\IncrementBy',
  120. 'incr' => '\Predis\Commands\Decrement',
  121. 'decrement' => '\Predis\Commands\Decrement',
  122. 'decrby' => '\Predis\Commands\DecrementBy',
  123. 'decrementBy' => '\Predis\Commands\DecrementBy',
  124. 'exists' => '\Predis\Commands\Exists',
  125. 'del' => '\Predis\Commands\Delete',
  126. 'delete' => '\Predis\Commands\Delete',
  127. 'type' => '\Predis\Commands\Type',
  128. /* commands operating on the key space */
  129. 'keys' => '\Predis\Commands\Keys',
  130. 'randomkey' => '\Predis\Commands\RandomKey',
  131. 'randomKey' => '\Predis\Commands\RandomKey',
  132. 'rename' => '\Predis\Commands\Rename',
  133. 'renamenx' => '\Predis\Commands\RenamePreserve',
  134. 'renamePreserve' => '\Predis\Commands\RenamePreserve',
  135. 'expire' => '\Predis\Commands\Expire',
  136. 'expireat' => '\Predis\Commands\ExpireAt',
  137. 'expireAt' => '\Predis\Commands\ExpireAt',
  138. 'dbsize' => '\Predis\Commands\DatabaseSize',
  139. 'databaseSize' => '\Predis\Commands\DatabaseSize',
  140. 'ttl' => '\Predis\Commands\TimeToLive',
  141. 'timeToLive' => '\Predis\Commands\TimeToLive',
  142. /* commands operating on lists */
  143. 'rpush' => '\Predis\Commands\ListPushTail',
  144. 'pushTail' => '\Predis\Commands\ListPushTail',
  145. 'lpush' => '\Predis\Commands\ListPushHead',
  146. 'pushHead' => '\Predis\Commands\ListPushHead',
  147. 'llen' => '\Predis\Commands\ListLength',
  148. 'listLength' => '\Predis\Commands\ListLength',
  149. 'lrange' => '\Predis\Commands\ListRange',
  150. 'listRange' => '\Predis\Commands\ListRange',
  151. 'ltrim' => '\Predis\Commands\ListTrim',
  152. 'listTrim' => '\Predis\Commands\ListTrim',
  153. 'lindex' => '\Predis\Commands\ListIndex',
  154. 'listIndex' => '\Predis\Commands\ListIndex',
  155. 'lset' => '\Predis\Commands\ListSet',
  156. 'listSet' => '\Predis\Commands\ListSet',
  157. 'lrem' => '\Predis\Commands\ListRemove',
  158. 'listRemove' => '\Predis\Commands\ListRemove',
  159. 'lpop' => '\Predis\Commands\ListPopFirst',
  160. 'popFirst' => '\Predis\Commands\ListPopFirst',
  161. 'rpop' => '\Predis\Commands\ListPopLast',
  162. 'popLast' => '\Predis\Commands\ListPopLast',
  163. /* commands operating on sets */
  164. 'sadd' => '\Predis\Commands\SetAdd',
  165. 'setAdd' => '\Predis\Commands\SetAdd',
  166. 'srem' => '\Predis\Commands\SetRemove',
  167. 'setRemove' => '\Predis\Commands\SetRemove',
  168. 'spop' => '\Predis\Commands\SetPop',
  169. 'setPop' => '\Predis\Commands\SetPop',
  170. 'smove' => '\Predis\Commands\SetMove',
  171. 'setMove' => '\Predis\Commands\SetMove',
  172. 'scard' => '\Predis\Commands\SetCardinality',
  173. 'setCardinality' => '\Predis\Commands\SetCardinality',
  174. 'sismember' => '\Predis\Commands\SetIsMember',
  175. 'setIsMember' => '\Predis\Commands\SetIsMember',
  176. 'sinter' => '\Predis\Commands\SetIntersection',
  177. 'setIntersection' => '\Predis\Commands\SetIntersection',
  178. 'sinterstore' => '\Predis\Commands\SetIntersectionStore',
  179. 'setIntersectionStore' => '\Predis\Commands\SetIntersectionStore',
  180. 'sunion' => '\Predis\Commands\SetUnion',
  181. 'setUnion' => '\Predis\Commands\SetUnion',
  182. 'sunionstore' => '\Predis\Commands\SetUnionStore',
  183. 'setUnionStore' => '\Predis\Commands\SetUnionStore',
  184. 'sdiff' => '\Predis\Commands\SetDifference',
  185. 'setDifference' => '\Predis\Commands\SetDifference',
  186. 'sdiffstore' => '\Predis\Commands\SetDifferenceStore',
  187. 'setDifferenceStore' => '\Predis\Commands\SetDifferenceStore',
  188. 'smembers' => '\Predis\Commands\SetMembers',
  189. 'setMembers' => '\Predis\Commands\SetMembers',
  190. 'srandmember' => '\Predis\Commands\SetRandomMember',
  191. 'setRandomMember' => '\Predis\Commands\SetRandomMember',
  192. /* commands operating on sorted sets */
  193. 'zadd' => '\Predis\Commands\ZSetAdd',
  194. 'zsetAdd' => '\Predis\Commands\ZSetAdd',
  195. 'zrem' => '\Predis\Commands\ZSetRemove',
  196. 'zsetRemove' => '\Predis\Commands\ZSetRemove',
  197. 'zrange' => '\Predis\Commands\ZSetRange',
  198. 'zsetRange' => '\Predis\Commands\ZSetRange',
  199. 'zrevrange' => '\Predis\Commands\ZSetReverseRange',
  200. 'zsetReverseRange' => '\Predis\Commands\ZSetReverseRange',
  201. 'zrangebyscore' => '\Predis\Commands\ZSetRangeByScore',
  202. 'zsetRangeByScore' => '\Predis\Commands\ZSetRangeByScore',
  203. 'zcard' => '\Predis\Commands\ZSetCardinality',
  204. 'zsetCardinality' => '\Predis\Commands\ZSetCardinality',
  205. 'zscore' => '\Predis\Commands\ZSetScore',
  206. 'zsetScore' => '\Predis\Commands\ZSetScore',
  207. 'zremrangebyscore' => '\Predis\Commands\ZSetRemoveRangeByScore',
  208. 'zsetRemoveRangeByScore' => '\Predis\Commands\ZSetRemoveRangeByScore',
  209. /* multiple databases handling commands */
  210. 'select' => '\Predis\Commands\SelectDatabase',
  211. 'selectDatabase' => '\Predis\Commands\SelectDatabase',
  212. 'move' => '\Predis\Commands\MoveKey',
  213. 'moveKey' => '\Predis\Commands\MoveKey',
  214. 'flushdb' => '\Predis\Commands\FlushDatabase',
  215. 'flushDatabase' => '\Predis\Commands\FlushDatabase',
  216. 'flushall' => '\Predis\Commands\FlushAll',
  217. 'flushDatabases' => '\Predis\Commands\FlushAll',
  218. /* sorting */
  219. 'sort' => '\Predis\Commands\Sort',
  220. /* remote server control commands */
  221. 'info' => '\Predis\Commands\Info',
  222. 'slaveof' => '\Predis\Commands\SlaveOf',
  223. 'slaveOf' => '\Predis\Commands\SlaveOf',
  224. /* persistence control commands */
  225. 'save' => '\Predis\Commands\Save',
  226. 'bgsave' => '\Predis\Commands\BackgroundSave',
  227. 'backgroundSave' => '\Predis\Commands\BackgroundSave',
  228. 'lastsave' => '\Predis\Commands\LastSave',
  229. 'lastSave' => '\Predis\Commands\LastSave',
  230. 'shutdown' => '\Predis\Commands\Shutdown'
  231. );
  232. }
  233. }
  234. /* ------------------------------------------------------------------------- */
  235. abstract class Command {
  236. private $_arguments;
  237. public abstract function getCommandId();
  238. public abstract function serializeRequest($command, $arguments);
  239. public function canBeHashed() {
  240. return true;
  241. }
  242. public function closesConnection() {
  243. return false;
  244. }
  245. protected function filterArguments(Array $arguments) {
  246. return $arguments;
  247. }
  248. public function setArguments(/* arguments */) {
  249. $this->_arguments = $this->filterArguments(func_get_args());
  250. }
  251. public function setArgumentsArray(Array $arguments) {
  252. $this->_arguments = $this->filterArguments($arguments);
  253. }
  254. protected function getArguments() {
  255. return $this->_arguments !== null ? $this->_arguments : array();
  256. }
  257. public function getArgument($index = 0) {
  258. return $this->_arguments !== null ? $this->_arguments[$index] : null;
  259. }
  260. public function parseResponse($data) {
  261. return $data;
  262. }
  263. public final function __invoke() {
  264. return $this->serializeRequest($this->getCommandId(), $this->getArguments());
  265. }
  266. }
  267. abstract class InlineCommand extends Command {
  268. public function serializeRequest($command, $arguments) {
  269. if (isset($arguments[0]) && is_array($arguments[0])) {
  270. $arguments[0] = implode($arguments[0], ' ');
  271. }
  272. return $command . ' ' . implode($arguments, ' ') . Response::NEWLINE;
  273. }
  274. }
  275. abstract class BulkCommand extends Command {
  276. public function serializeRequest($command, $arguments) {
  277. $data = array_pop($arguments);
  278. if (is_array($data)) {
  279. $data = implode($data, ' ');
  280. }
  281. return $command . ' ' . implode($arguments, ' ') . ' ' . strlen($data) .
  282. Response::NEWLINE . $data . Response::NEWLINE;
  283. }
  284. }
  285. abstract class MultiBulkCommand extends Command {
  286. public function serializeRequest($command, $arguments) {
  287. $buffer = array();
  288. $cmd_args = null;
  289. if (count($arguments) === 1 && is_array($arguments[0])) {
  290. $cmd_args = array();
  291. foreach ($arguments[0] as $k => $v) {
  292. $cmd_args[] = $k;
  293. $cmd_args[] = $v;
  294. }
  295. }
  296. else {
  297. $cmd_args = $arguments;
  298. }
  299. $buffer[] = '*' . ((string) count($cmd_args) + 1) . Response::NEWLINE;
  300. $buffer[] = '$' . strlen($command) . Response::NEWLINE . $command . Response::NEWLINE;
  301. foreach ($cmd_args as $argument) {
  302. $buffer[] = '$' . strlen($argument) . Response::NEWLINE . $argument . Response::NEWLINE;
  303. }
  304. return implode('', $buffer);
  305. }
  306. }
  307. /* ------------------------------------------------------------------------- */
  308. class Response {
  309. const NEWLINE = "\r\n";
  310. const OK = 'OK';
  311. const ERROR = 'ERR';
  312. const NULL = 'nil';
  313. private static $_prefixHandlers;
  314. private static function initializePrefixHandlers() {
  315. return array(
  316. // status
  317. '+' => function($socket) {
  318. $status = rtrim(fgets($socket), Response::NEWLINE);
  319. return $status === Response::OK ? true : $status;
  320. },
  321. // error
  322. '-' => function($socket) {
  323. $errorMessage = rtrim(fgets($socket), Response::NEWLINE);
  324. throw new ServerException(substr($errorMessage, 4));
  325. },
  326. // bulk
  327. '$' => function($socket) {
  328. $dataLength = rtrim(fgets($socket), Response::NEWLINE);
  329. if (!is_numeric($dataLength)) {
  330. throw new ClientException("Cannot parse '$dataLength' as data length");
  331. }
  332. if ($dataLength > 0) {
  333. $value = fread($socket, $dataLength);
  334. fread($socket, 2);
  335. return $value;
  336. }
  337. else if ($dataLength == 0) {
  338. // TODO: I just have a doubt here...
  339. fread($socket, 2);
  340. }
  341. return null;
  342. },
  343. // multibulk
  344. '*' => function($socket) {
  345. $rawLength = rtrim(fgets($socket), Response::NEWLINE);
  346. if (!is_numeric($rawLength)) {
  347. throw new ClientException("Cannot parse '$rawLength' as data length");
  348. }
  349. $listLength = (int) $rawLength;
  350. if ($listLength === -1) {
  351. return null;
  352. }
  353. $list = array();
  354. if ($listLength > 0) {
  355. for ($i = 0; $i < $listLength; $i++) {
  356. $handler = Response::getPrefixHandler(fgetc($socket));
  357. $list[] = $handler($socket);
  358. }
  359. }
  360. return $list;
  361. },
  362. // integer
  363. ':' => function($socket) {
  364. $number = rtrim(fgets($socket), Response::NEWLINE);
  365. if (is_numeric($number)) {
  366. return (int) $number;
  367. }
  368. else {
  369. if ($number !== Response::NULL) {
  370. throw new ClientException("Cannot parse '$number' as numeric response");
  371. }
  372. return null;
  373. }
  374. }
  375. );
  376. }
  377. public static function getPrefixHandler($prefix) {
  378. if (self::$_prefixHandlers == null) {
  379. self::$_prefixHandlers = self::initializePrefixHandlers();
  380. }
  381. $handler = self::$_prefixHandlers[$prefix];
  382. if ($handler === null) {
  383. throw new MalformedServerResponse("Unknown prefix '$prefix'");
  384. }
  385. return $handler;
  386. }
  387. }
  388. class CommandPipeline {
  389. private $_redisClient, $_pipelineBuffer, $_returnValues, $_running;
  390. public function __construct(Client $redisClient) {
  391. $this->_redisClient = $redisClient;
  392. $this->_pipelineBuffer = array();
  393. $this->_returnValues = array();
  394. }
  395. public function __call($method, $arguments) {
  396. $command = $this->_redisClient->createCommandInstance($method, $arguments);
  397. $this->recordCommand($command);
  398. }
  399. private function recordCommand(Command $command) {
  400. $this->_pipelineBuffer[] = $command;
  401. }
  402. private function getRecordedCommands() {
  403. return $this->_pipelineBuffer;
  404. }
  405. public function flushPipeline() {
  406. if (count($this->_pipelineBuffer) === 0) {
  407. return;
  408. }
  409. $connection = $this->_redisClient->getConnection();
  410. $commands = &$this->getRecordedCommands();
  411. foreach ($commands as $command) {
  412. $connection->writeCommand($command);
  413. }
  414. foreach ($commands as $command) {
  415. $this->_returnValues[] = $connection->readResponse($command);
  416. }
  417. $this->_pipelineBuffer = array();
  418. }
  419. private function setRunning($bool) {
  420. // TODO: I am honest when I say that I don't like this approach.
  421. if ($bool == true && $this->_running == true) {
  422. throw new ClientException("This pipeline is already opened");
  423. }
  424. $this->_running = $bool;
  425. }
  426. public function execute(\Closure $block = null) {
  427. $this->setRunning(true);
  428. $pipelineBlockException = null;
  429. try {
  430. if ($block !== null) {
  431. $block($this);
  432. }
  433. $this->flushPipeline();
  434. }
  435. catch (\Exception $exception) {
  436. $pipelineBlockException = $exception;
  437. }
  438. $this->setRunning(false);
  439. if ($pipelineBlockException !== null) {
  440. throw $pipelineBlockException;
  441. }
  442. return $this->_returnValues;
  443. }
  444. }
  445. /* ------------------------------------------------------------------------- */
  446. interface IConnection {
  447. public function connect();
  448. public function disconnect();
  449. public function isConnected();
  450. public function writeCommand(Command $command);
  451. public function readResponse(Command $command);
  452. }
  453. class Connection implements IConnection {
  454. const DEFAULT_HOST = '127.0.0.1';
  455. const DEFAULT_PORT = 6379;
  456. const CONNECTION_TIMEOUT = 2;
  457. const READ_WRITE_TIMEOUT = 5;
  458. private $_host, $_port, $_socket;
  459. public function __construct($host = self::DEFAULT_HOST, $port = self::DEFAULT_PORT) {
  460. $this->_host = $host;
  461. $this->_port = $port;
  462. }
  463. public function __destruct() {
  464. $this->disconnect();
  465. }
  466. public function isConnected() {
  467. return is_resource($this->_socket);
  468. }
  469. public function connect() {
  470. if ($this->isConnected()) {
  471. throw new ClientException('Connection already estabilished');
  472. }
  473. $uri = sprintf('tcp://%s:%d/', $this->_host, $this->_port);
  474. $this->_socket = @stream_socket_client($uri, $errno, $errstr, self::CONNECTION_TIMEOUT);
  475. if (!$this->_socket) {
  476. throw new ClientException(trim($errstr), $errno);
  477. }
  478. stream_set_timeout($this->_socket, self::READ_WRITE_TIMEOUT);
  479. }
  480. public function disconnect() {
  481. if ($this->isConnected()) {
  482. fclose($this->_socket);
  483. }
  484. }
  485. public function writeCommand(Command $command) {
  486. fwrite($this->getSocket(), $command());
  487. }
  488. public function readResponse(Command $command) {
  489. $socket = $this->getSocket();
  490. $handler = Response::getPrefixHandler(fgetc($socket));
  491. $response = $command->parseResponse($handler($socket));
  492. return $response;
  493. }
  494. public function rawCommand($rawCommandData, $closesConnection = false) {
  495. $socket = $this->getSocket();
  496. fwrite($socket, $rawCommandData);
  497. if ($closesConnection) {
  498. return;
  499. }
  500. $handler = Response::getPrefixHandler(fgetc($socket));
  501. return $handler($socket);
  502. }
  503. public function getSocket() {
  504. if (!$this->isConnected()) {
  505. $this->connect();
  506. }
  507. return $this->_socket;
  508. }
  509. public function __toString() {
  510. return sprintf('tcp://%s:%d/', $this->_host, $this->_port);
  511. }
  512. }
  513. class ConnectionCluster implements IConnection {
  514. // TODO: storing a temporary map of commands hashes to hashring items (that
  515. // is, connections) could offer a notable speedup, but I am wondering
  516. // about the increased memory footprint.
  517. // TODO: find a clean way to handle connection failures of single nodes.
  518. private $_pool, $_ring;
  519. public function __construct() {
  520. $this->_pool = array();
  521. $this->_ring = new Utilities\HashRing();
  522. }
  523. public function __destruct() {
  524. $this->disconnect();
  525. }
  526. public function isConnected() {
  527. foreach ($this->_pool as $connection) {
  528. if ($connection->isConnected()) {
  529. return true;
  530. }
  531. }
  532. return false;
  533. }
  534. public function connect() {
  535. foreach ($this->_pool as $connection) {
  536. $connection->connect();
  537. }
  538. }
  539. public function disconnect() {
  540. foreach ($this->_pool as $connection) {
  541. $connection->disconnect();
  542. }
  543. }
  544. public function add(Connection $connection) {
  545. $this->_pool[] = $connection;
  546. $this->_ring->add($connection);
  547. }
  548. private function getConnectionFromRing(Command $command) {
  549. return $this->_ring->get($this->computeHash($command));
  550. }
  551. private function computeHash(Command $command) {
  552. return crc32($command->getArgument(0));
  553. }
  554. private function getConnection(Command $command) {
  555. return $command->canBeHashed()
  556. ? $this->getConnectionFromRing($command)
  557. : $this->getConnectionById(0);
  558. }
  559. public function getConnectionById($id = null) {
  560. return $this->_pool[$id === null ? 0 : $id];
  561. }
  562. public function writeCommand(Command $command) {
  563. $this->getConnection($command)->writeCommand($command);
  564. }
  565. public function readResponse(Command $command) {
  566. return $this->getConnection($command)->readResponse($command);
  567. }
  568. }
  569. /* ------------------------------------------------------------------------- */
  570. namespace Predis\Utilities;
  571. class HashRing {
  572. const NUMBER_OF_REPLICAS = 64;
  573. private $_ring, $_ringKeys;
  574. public function __construct() {
  575. $this->_ring = array();
  576. $this->_ringKeys = array();
  577. }
  578. public function add($node) {
  579. for ($i = 0; $i < self::NUMBER_OF_REPLICAS; $i++) {
  580. $key = crc32((string)$node . ':' . $i);
  581. $this->_ring[$key] = $node;
  582. }
  583. ksort($this->_ring, SORT_NUMERIC);
  584. $this->_ringKeys = array_keys($this->_ring);
  585. }
  586. public function remove($node) {
  587. for ($i = 0; $i < self::NUMBER_OF_REPLICAS; $i++) {
  588. $key = crc32((string)$node . '_' . $i);
  589. unset($this->_ring[$key]);
  590. $this->_ringKeys = array_filter($this->_ringKeys, function($rk) use($key) {
  591. return $rk !== $key;
  592. });
  593. }
  594. }
  595. public function get($key) {
  596. return $this->_ring[$this->getNodeKey($key)];
  597. }
  598. private function getNodeKey($key) {
  599. $upper = count($this->_ringKeys) - 1;
  600. $lower = 0;
  601. $index = 0;
  602. while ($lower <= $upper) {
  603. $index = ($lower + $upper) / 2;
  604. $item = $this->_ringKeys[$index];
  605. if ($item === $key) {
  606. return $index;
  607. }
  608. else if ($item > $key) {
  609. $upper = $index - 1;
  610. }
  611. else {
  612. $lower = $index + 1;
  613. }
  614. }
  615. return $this->_ringKeys[$upper];
  616. }
  617. }
  618. /* ------------------------------------------------------------------------- */
  619. namespace Predis\Commands;
  620. /* miscellaneous commands */
  621. class Ping extends \Predis\InlineCommand {
  622. public function canBeHashed() { return false; }
  623. public function getCommandId() { return 'PING'; }
  624. public function parseResponse($data) {
  625. return $data === 'PONG' ? true : false;
  626. }
  627. }
  628. class DoEcho extends \Predis\BulkCommand {
  629. public function canBeHashed() { return false; }
  630. public function getCommandId() { return 'ECHO'; }
  631. }
  632. class Auth extends \Predis\InlineCommand {
  633. public function canBeHashed() { return false; }
  634. public function getCommandId() { return 'AUTH'; }
  635. }
  636. /* connection handling */
  637. class Quit extends \Predis\InlineCommand {
  638. public function canBeHashed() { return false; }
  639. public function getCommandId() { return 'QUIT'; }
  640. public function closesConnection() { return true; }
  641. }
  642. /* commands operating on string values */
  643. class Set extends \Predis\BulkCommand {
  644. public function getCommandId() { return 'SET'; }
  645. }
  646. class SetPreserve extends \Predis\BulkCommand {
  647. public function getCommandId() { return 'SETNX'; }
  648. public function parseResponse($data) { return (bool) $data; }
  649. }
  650. class SetMultiple extends \Predis\MultiBulkCommand {
  651. public function canBeHashed() { return false; }
  652. public function getCommandId() { return 'MSET'; }
  653. }
  654. class SetMultiplePreserve extends \Predis\MultiBulkCommand {
  655. public function canBeHashed() { return false; }
  656. public function getCommandId() { return 'MSETNX'; }
  657. public function parseResponse($data) { return (bool) $data; }
  658. }
  659. class Get extends \Predis\InlineCommand {
  660. public function getCommandId() { return 'GET'; }
  661. }
  662. class GetMultiple extends \Predis\InlineCommand {
  663. public function canBeHashed() { return false; }
  664. public function getCommandId() { return 'MGET'; }
  665. }
  666. class GetSet extends \Predis\BulkCommand {
  667. public function getCommandId() { return 'GETSET'; }
  668. }
  669. class Increment extends \Predis\InlineCommand {
  670. public function getCommandId() { return 'INCR'; }
  671. }
  672. class IncrementBy extends \Predis\InlineCommand {
  673. public function getCommandId() { return 'INCRBY'; }
  674. }
  675. class Decrement extends \Predis\InlineCommand {
  676. public function getCommandId() { return 'DECR'; }
  677. }
  678. class DecrementBy extends \Predis\InlineCommand {
  679. public function getCommandId() { return 'DECRBY'; }
  680. }
  681. class Exists extends \Predis\InlineCommand {
  682. public function getCommandId() { return 'EXISTS'; }
  683. public function parseResponse($data) { return (bool) $data; }
  684. }
  685. class Delete extends \Predis\InlineCommand {
  686. public function getCommandId() { return 'DEL'; }
  687. public function parseResponse($data) { return (bool) $data; }
  688. }
  689. class Type extends \Predis\InlineCommand {
  690. public function getCommandId() { return 'TYPE'; }
  691. }
  692. /* commands operating on the key space */
  693. class Keys extends \Predis\InlineCommand {
  694. public function canBeHashed() { return false; }
  695. public function getCommandId() { return 'KEYS'; }
  696. public function parseResponse($data) {
  697. // TODO: is this behaviour correct?
  698. return strlen($data) > 0 ? explode(' ', $data) : array();
  699. }
  700. }
  701. class RandomKey extends \Predis\InlineCommand {
  702. public function canBeHashed() { return false; }
  703. public function getCommandId() { return 'RANDOMKEY'; }
  704. public function parseResponse($data) { return $data !== '' ? $data : null; }
  705. }
  706. class Rename extends \Predis\InlineCommand {
  707. // TODO: doesn't RENAME break the hash-based client-side sharding?
  708. public function canBeHashed() { return false; }
  709. public function getCommandId() { return 'RENAME'; }
  710. }
  711. class RenamePreserve extends \Predis\InlineCommand {
  712. public function canBeHashed() { return false; }
  713. public function getCommandId() { return 'RENAMENX'; }
  714. public function parseResponse($data) { return (bool) $data; }
  715. }
  716. class Expire extends \Predis\InlineCommand {
  717. public function getCommandId() { return 'EXPIRE'; }
  718. public function parseResponse($data) { return (bool) $data; }
  719. }
  720. class ExpireAt extends \Predis\InlineCommand {
  721. public function getCommandId() { return 'EXPIREAT'; }
  722. public function parseResponse($data) { return (bool) $data; }
  723. }
  724. class DatabaseSize extends \Predis\InlineCommand {
  725. public function canBeHashed() { return false; }
  726. public function getCommandId() { return 'DBSIZE'; }
  727. }
  728. class TimeToLive extends \Predis\InlineCommand {
  729. public function getCommandId() { return 'TTL'; }
  730. }
  731. /* commands operating on lists */
  732. class ListPushTail extends \Predis\BulkCommand {
  733. public function getCommandId() { return 'RPUSH'; }
  734. }
  735. class ListPushHead extends \Predis\BulkCommand {
  736. public function getCommandId() { return 'LPUSH'; }
  737. }
  738. class ListLength extends \Predis\InlineCommand {
  739. public function getCommandId() { return 'LLEN'; }
  740. }
  741. class ListRange extends \Predis\InlineCommand {
  742. public function getCommandId() { return 'LRANGE'; }
  743. }
  744. class ListTrim extends \Predis\InlineCommand {
  745. public function getCommandId() { return 'LTRIM'; }
  746. }
  747. class ListIndex extends \Predis\InlineCommand {
  748. public function getCommandId() { return 'LINDEX'; }
  749. }
  750. class ListSet extends \Predis\BulkCommand {
  751. public function getCommandId() { return 'LSET'; }
  752. }
  753. class ListRemove extends \Predis\BulkCommand {
  754. public function getCommandId() { return 'LREM'; }
  755. }
  756. class ListPopFirst extends \Predis\InlineCommand {
  757. public function getCommandId() { return 'LPOP'; }
  758. }
  759. class ListPopLast extends \Predis\InlineCommand {
  760. public function getCommandId() { return 'RPOP'; }
  761. }
  762. /* commands operating on sets */
  763. class SetAdd extends \Predis\BulkCommand {
  764. public function getCommandId() { return 'SADD'; }
  765. public function parseResponse($data) { return (bool) $data; }
  766. }
  767. class SetRemove extends \Predis\BulkCommand {
  768. public function getCommandId() { return 'SREM'; }
  769. public function parseResponse($data) { return (bool) $data; }
  770. }
  771. class SetPop extends \Predis\InlineCommand {
  772. public function getCommandId() { return 'SPOP'; }
  773. }
  774. class SetMove extends \Predis\BulkCommand {
  775. public function canBeHashed() { return false; }
  776. public function getCommandId() { return 'SMOVE'; }
  777. public function parseResponse($data) { return (bool) $data; }
  778. }
  779. class SetCardinality extends \Predis\InlineCommand {
  780. public function getCommandId() { return 'SCARD'; }
  781. }
  782. class SetIsMember extends \Predis\BulkCommand {
  783. public function getCommandId() { return 'SISMEMBER'; }
  784. public function parseResponse($data) { return (bool) $data; }
  785. }
  786. class SetIntersection extends \Predis\InlineCommand {
  787. public function getCommandId() { return 'SINTER'; }
  788. }
  789. class SetIntersectionStore extends \Predis\InlineCommand {
  790. public function getCommandId() { return 'SINTERSTORE'; }
  791. }
  792. class SetUnion extends \Predis\InlineCommand {
  793. public function getCommandId() { return 'SUNION'; }
  794. }
  795. class SetUnionStore extends \Predis\InlineCommand {
  796. public function getCommandId() { return 'SUNIONSTORE'; }
  797. }
  798. class SetDifference extends \Predis\InlineCommand {
  799. public function getCommandId() { return 'SDIFF'; }
  800. }
  801. class SetDifferenceStore extends \Predis\InlineCommand {
  802. public function getCommandId() { return 'SDIFFSTORE'; }
  803. }
  804. class SetMembers extends \Predis\InlineCommand {
  805. public function getCommandId() { return 'SMEMBERS'; }
  806. }
  807. class SetRandomMember extends \Predis\InlineCommand {
  808. public function getCommandId() { return 'SRANDMEMBER'; }
  809. }
  810. /* commands operating on sorted sets */
  811. class ZSetAdd extends \Predis\BulkCommand {
  812. public function getCommandId() { return 'ZADD'; }
  813. public function parseResponse($data) { return (bool) $data; }
  814. }
  815. class ZSetRemove extends \Predis\BulkCommand {
  816. public function getCommandId() { return 'ZREM'; }
  817. public function parseResponse($data) { return (bool) $data; }
  818. }
  819. class ZSetRange extends \Predis\InlineCommand {
  820. public function getCommandId() { return 'ZRANGE'; }
  821. }
  822. class ZSetReverseRange extends \Predis\InlineCommand {
  823. public function getCommandId() { return 'ZREVRANGE'; }
  824. }
  825. class ZSetRangeByScore extends \Predis\InlineCommand {
  826. public function getCommandId() { return 'ZRANGEBYSCORE'; }
  827. }
  828. class ZSetCardinality extends \Predis\InlineCommand {
  829. public function getCommandId() { return 'ZCARD'; }
  830. }
  831. class ZSetScore extends \Predis\BulkCommand {
  832. public function getCommandId() { return 'ZSCORE'; }
  833. }
  834. class ZSetRemoveRangeByScore extends \Predis\InlineCommand {
  835. public function getCommandId() { return 'ZREMRANGEBYSCORE'; }
  836. }
  837. /* multiple databases handling commands */
  838. class SelectDatabase extends \Predis\InlineCommand {
  839. public function canBeHashed() { return false; }
  840. public function getCommandId() { return 'SELECT'; }
  841. }
  842. class MoveKey extends \Predis\InlineCommand {
  843. public function canBeHashed() { return false; }
  844. public function getCommandId() { return 'MOVE'; }
  845. public function parseResponse($data) { return (bool) $data; }
  846. }
  847. class FlushDatabase extends \Predis\InlineCommand {
  848. public function canBeHashed() { return false; }
  849. public function getCommandId() { return 'FLUSHDB'; }
  850. }
  851. class FlushAll extends \Predis\InlineCommand {
  852. public function canBeHashed() { return false; }
  853. public function getCommandId() { return 'FLUSHALL'; }
  854. }
  855. /* sorting */
  856. class Sort extends \Predis\InlineCommand {
  857. public function getCommandId() { return 'SORT'; }
  858. public function filterArguments($arguments) {
  859. if (count($arguments) === 1) {
  860. return $arguments;
  861. }
  862. // TODO: add more parameters checks
  863. $query = array($arguments[0]);
  864. $sortParams = $arguments[1];
  865. if (isset($sortParams['by'])) {
  866. $query[] = 'BY ' . $sortParams['by'];
  867. }
  868. if (isset($sortParams['get'])) {
  869. $query[] = 'GET ' . $sortParams['get'];
  870. }
  871. if (isset($sortParams['limit']) && is_array($sortParams['limit'])) {
  872. $query[] = 'LIMIT ' . $sortParams['limit'][0] . ' ' . $sortParams['limit'][1];
  873. }
  874. if (isset($sortParams['sort'])) {
  875. $query[] = strtoupper($sortParams['sort']);
  876. }
  877. if (isset($sortParams['alpha']) && $sortParams['alpha'] == true) {
  878. $query[] = 'ALPHA';
  879. }
  880. if (isset($sortParams['store']) && $sortParams['store'] == true) {
  881. $query[] = 'STORE ' . $sortParams['store'];
  882. }
  883. return $query;
  884. }
  885. }
  886. /* persistence control commands */
  887. class Save extends \Predis\InlineCommand {
  888. public function canBeHashed() { return false; }
  889. public function getCommandId() { return 'SAVE'; }
  890. }
  891. class BackgroundSave extends \Predis\InlineCommand {
  892. public function canBeHashed() { return false; }
  893. public function getCommandId() { return 'BGSAVE'; }
  894. }
  895. class LastSave extends \Predis\InlineCommand {
  896. public function canBeHashed() { return false; }
  897. public function getCommandId() { return 'LASTSAVE'; }
  898. }
  899. class Shutdown extends \Predis\InlineCommand {
  900. public function canBeHashed() { return false; }
  901. public function getCommandId() { return 'SHUTDOWN'; }
  902. public function closesConnection() { return true; }
  903. }
  904. /* remote server control commands */
  905. class Info extends \Predis\InlineCommand {
  906. public function canBeHashed() { return false; }
  907. public function getCommandId() { return 'INFO'; }
  908. public function parseResponse($data) {
  909. $info = array();
  910. $infoLines = explode("\r\n", $data, -1);
  911. foreach ($infoLines as $row) {
  912. list($k, $v) = explode(':', $row);
  913. $info[$k] = $v;
  914. }
  915. return $info;
  916. }
  917. }
  918. class SlaveOf extends \Predis\InlineCommand {
  919. public function canBeHashed() { return false; }
  920. public function getCommandId() { return 'SLAVEOF'; }
  921. public function filterArguments($arguments) {
  922. return count($arguments) === 0 ? array('NO ONE') : $arguments;
  923. }
  924. }
  925. ?>