RedisTrait.php 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788
  1. <?php
  2. /*
  3. * This file is part of the Symfony package.
  4. *
  5. * (c) Fabien Potencier <fabien@symfony.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 Symfony\Component\Cache\Traits;
  11. use Predis\Command\Redis\UNLINK;
  12. use Predis\Connection\Aggregate\ClusterInterface;
  13. use Predis\Connection\Aggregate\RedisCluster;
  14. use Predis\Connection\Aggregate\ReplicationInterface;
  15. use Predis\Connection\Cluster\ClusterInterface as Predis2ClusterInterface;
  16. use Predis\Connection\Cluster\RedisCluster as Predis2RedisCluster;
  17. use Predis\Connection\Replication\ReplicationInterface as Predis2ReplicationInterface;
  18. use Predis\Response\ErrorInterface;
  19. use Predis\Response\Status;
  20. use Relay\Cluster as RelayCluster;
  21. use Relay\Relay;
  22. use Relay\Sentinel;
  23. use Symfony\Component\Cache\Exception\CacheException;
  24. use Symfony\Component\Cache\Exception\InvalidArgumentException;
  25. use Symfony\Component\Cache\Marshaller\DefaultMarshaller;
  26. use Symfony\Component\Cache\Marshaller\MarshallerInterface;
  27. /**
  28. * @author Aurimas Niekis <aurimas@niekis.lt>
  29. * @author Nicolas Grekas <p@tchwork.com>
  30. *
  31. * @internal
  32. */
  33. trait RedisTrait
  34. {
  35. private static array $defaultConnectionOptions = [
  36. 'class' => null,
  37. 'auth' => null,
  38. 'persistent' => false,
  39. 'persistent_id' => null,
  40. 'timeout' => 30,
  41. 'read_timeout' => 0,
  42. 'retry_interval' => 0,
  43. 'tcp_keepalive' => 0,
  44. 'lazy' => null,
  45. 'cluster' => false,
  46. 'cluster_command_timeout' => 0,
  47. 'cluster_relay_context' => [],
  48. 'sentinel' => null,
  49. 'dbindex' => 0,
  50. 'failover' => 'none',
  51. 'ssl' => null, // see https://php.net/context.ssl
  52. ];
  53. private \Redis|Relay|RelayCluster|\RedisArray|\RedisCluster|\Predis\ClientInterface $redis;
  54. private MarshallerInterface $marshaller;
  55. private function init(\Redis|Relay|RelayCluster|\RedisArray|\RedisCluster|\Predis\ClientInterface $redis, string $namespace, int $defaultLifetime, ?MarshallerInterface $marshaller): void
  56. {
  57. parent::__construct($namespace, $defaultLifetime);
  58. if (preg_match('#[^-+_.A-Za-z0-9]#', $namespace, $match)) {
  59. throw new InvalidArgumentException(\sprintf('RedisAdapter namespace contains "%s" but only characters in [-+_.A-Za-z0-9] are allowed.', $match[0]));
  60. }
  61. if ($redis instanceof \Predis\ClientInterface && $redis->getOptions()->exceptions) {
  62. $options = clone $redis->getOptions();
  63. \Closure::bind(function () { $this->options['exceptions'] = false; }, $options, $options)();
  64. $redis = new $redis($redis->getConnection(), $options);
  65. }
  66. $this->redis = $redis;
  67. $this->marshaller = $marshaller ?? new DefaultMarshaller();
  68. }
  69. /**
  70. * Creates a Redis connection using a DSN configuration.
  71. *
  72. * Example DSN:
  73. * - redis://localhost
  74. * - redis://example.com:1234
  75. * - redis://secret@example.com/13
  76. * - redis:///var/run/redis.sock
  77. * - redis://secret@/var/run/redis.sock/13
  78. *
  79. * @param array $options See self::$defaultConnectionOptions
  80. *
  81. * @throws InvalidArgumentException when the DSN is invalid
  82. */
  83. public static function createConnection(#[\SensitiveParameter] string $dsn, array $options = []): \Redis|\RedisArray|\RedisCluster|\Predis\ClientInterface|Relay|RelayCluster
  84. {
  85. $scheme = match (true) {
  86. str_starts_with($dsn, 'redis:') => 'redis',
  87. str_starts_with($dsn, 'rediss:') => 'rediss',
  88. str_starts_with($dsn, 'valkey:') => 'valkey',
  89. str_starts_with($dsn, 'valkeys:') => 'valkeys',
  90. default => throw new InvalidArgumentException('Invalid Redis DSN: it does not start with "redis[s]:" nor "valkey[s]:".'),
  91. };
  92. if (!\extension_loaded('redis') && !\extension_loaded('relay') && !class_exists(\Predis\Client::class)) {
  93. throw new CacheException('Cannot find the "redis" extension nor the "relay" extension nor the "predis/predis" package.');
  94. }
  95. $auth = null;
  96. $params = preg_replace_callback('#^'.$scheme.':(//)?(?:(?:(?<user>[^:@]*+):)?(?<password>[^@]*+)@)?#', function ($m) use (&$auth) {
  97. if (isset($m['password'])) {
  98. if (\in_array($m['user'], ['', 'default'], true)) {
  99. $auth = rawurldecode($m['password']);
  100. } else {
  101. $auth = [rawurldecode($m['user']), rawurldecode($m['password'])];
  102. }
  103. if ('' === $auth) {
  104. $auth = null;
  105. }
  106. }
  107. return 'file:'.($m[1] ?? '');
  108. }, $dsn);
  109. if (false === $params = parse_url($params)) {
  110. throw new InvalidArgumentException('Invalid Redis DSN.');
  111. }
  112. $query = $hosts = [];
  113. $tls = 'rediss' === $scheme || 'valkeys' === $scheme;
  114. $tcpScheme = $tls ? 'tls' : 'tcp';
  115. if (isset($params['query'])) {
  116. parse_str($params['query'], $query);
  117. if (isset($query['host'])) {
  118. if (!\is_array($hosts = $query['host'])) {
  119. throw new InvalidArgumentException('Invalid Redis DSN: query parameter "host" must be an array.');
  120. }
  121. foreach ($hosts as $host => $parameters) {
  122. if (\is_string($parameters)) {
  123. parse_str($parameters, $parameters);
  124. }
  125. if (false === $i = strrpos($host, ':')) {
  126. $hosts[$host] = ['scheme' => $tcpScheme, 'host' => $host, 'port' => 6379] + $parameters;
  127. } elseif ($port = (int) substr($host, 1 + $i)) {
  128. $hosts[$host] = ['scheme' => $tcpScheme, 'host' => substr($host, 0, $i), 'port' => $port] + $parameters;
  129. } else {
  130. $hosts[$host] = ['scheme' => 'unix', 'path' => substr($host, 0, $i)] + $parameters;
  131. }
  132. }
  133. $hosts = array_values($hosts);
  134. }
  135. }
  136. if (isset($params['host']) || isset($params['path'])) {
  137. if (!isset($params['dbindex']) && isset($params['path'])) {
  138. if (preg_match('#/(\d+)?$#', $params['path'], $m)) {
  139. $params['dbindex'] = $m[1] ?? $query['dbindex'] ?? '0';
  140. $params['path'] = substr($params['path'], 0, -\strlen($m[0]));
  141. } elseif (isset($params['host'])) {
  142. throw new InvalidArgumentException('Invalid Redis DSN: parameter "dbindex" must be a number.');
  143. }
  144. }
  145. if (isset($params['host'])) {
  146. array_unshift($hosts, ['scheme' => $tcpScheme, 'host' => $params['host'], 'port' => $params['port'] ?? 6379]);
  147. } else {
  148. array_unshift($hosts, ['scheme' => 'unix', 'path' => $params['path']]);
  149. }
  150. }
  151. if (!$hosts) {
  152. throw new InvalidArgumentException('Invalid Redis DSN: missing host.');
  153. }
  154. if (isset($params['dbindex'], $query['dbindex']) && $params['dbindex'] !== $query['dbindex']) {
  155. throw new InvalidArgumentException('Invalid Redis DSN: path and query "dbindex" parameters mismatch.');
  156. }
  157. $params += $query + $options + self::$defaultConnectionOptions;
  158. $params['auth'] ??= $auth;
  159. $aliases = [
  160. 'sentinel_master' => 'sentinel',
  161. 'redis_sentinel' => 'sentinel',
  162. 'redis_cluster' => 'cluster',
  163. ];
  164. foreach ($aliases as $alias => $key) {
  165. $params[$key] = match (true) {
  166. \array_key_exists($key, $query) => $query[$key],
  167. \array_key_exists($alias, $query) => $query[$alias],
  168. \array_key_exists($key, $options) => $options[$key],
  169. \array_key_exists($alias, $options) => $options[$alias],
  170. default => $params[$key],
  171. };
  172. }
  173. if (isset($params['sentinel']) && !class_exists(\Predis\Client::class) && !class_exists(\RedisSentinel::class) && !class_exists(Sentinel::class)) {
  174. throw new CacheException('Redis Sentinel support requires one of: "predis/predis", "ext-redis >= 5.2", "ext-relay".');
  175. }
  176. foreach (['lazy', 'persistent', 'cluster'] as $option) {
  177. if (!\is_bool($params[$option] ?? false)) {
  178. $params[$option] = filter_var($params[$option], \FILTER_VALIDATE_BOOLEAN);
  179. }
  180. }
  181. if ($params['cluster'] && isset($params['sentinel'])) {
  182. throw new InvalidArgumentException('Cannot use both "cluster" and "sentinel" at the same time.');
  183. }
  184. $class = $params['class'] ?? match (true) {
  185. $params['cluster'] => match (true) {
  186. \extension_loaded('redis') => \RedisCluster::class,
  187. \extension_loaded('relay') => RelayCluster::class,
  188. default => \Predis\Client::class,
  189. },
  190. isset($params['sentinel']) => match (true) {
  191. \extension_loaded('redis') => \Redis::class,
  192. \extension_loaded('relay') => Relay::class,
  193. default => \Predis\Client::class,
  194. },
  195. 1 < \count($hosts) && \extension_loaded('redis') => \RedisArray::class,
  196. \extension_loaded('redis') => \Redis::class,
  197. \extension_loaded('relay') => Relay::class,
  198. default => \Predis\Client::class,
  199. };
  200. if (isset($params['sentinel']) && !is_a($class, \Predis\Client::class, true) && !class_exists(\RedisSentinel::class) && !class_exists(Sentinel::class)) {
  201. throw new CacheException(\sprintf('Cannot use Redis Sentinel: class "%s" does not extend "Predis\Client" and neither ext-redis >= 5.2 nor ext-relay have been found.', $class));
  202. }
  203. $isRedisExt = is_a($class, \Redis::class, true);
  204. $isRelayExt = !$isRedisExt && is_a($class, Relay::class, true);
  205. if ($isRedisExt || $isRelayExt) {
  206. $connect = $params['persistent'] || $params['persistent_id'] ? 'pconnect' : 'connect';
  207. $initializer = static function () use ($class, $isRedisExt, $connect, $params, $auth, $hosts, $tls) {
  208. $sentinelClass = $isRedisExt ? \RedisSentinel::class : Sentinel::class;
  209. $redis = new $class();
  210. $hostIndex = 0;
  211. do {
  212. $host = $hosts[$hostIndex]['host'] ?? $hosts[$hostIndex]['path'];
  213. $port = $hosts[$hostIndex]['port'] ?? 0;
  214. $passAuth = null !== $params['auth'] && (!$isRedisExt || \defined('Redis::OPT_NULL_MULTIBULK_AS_NULL'));
  215. $address = false;
  216. if (isset($hosts[$hostIndex]['host']) && $tls) {
  217. $host = 'tls://'.$host;
  218. }
  219. if (!isset($params['sentinel'])) {
  220. break;
  221. }
  222. try {
  223. if ($isRedisExt && version_compare(phpversion('redis'), '6.0.0', '>=')) {
  224. $options = [
  225. 'host' => $host,
  226. 'port' => $port,
  227. 'connectTimeout' => (float) $params['timeout'],
  228. 'persistent' => $params['persistent_id'],
  229. 'retryInterval' => (int) $params['retry_interval'],
  230. 'readTimeout' => (float) $params['read_timeout'],
  231. ];
  232. if ($passAuth) {
  233. $options['auth'] = $params['auth'];
  234. }
  235. $sentinel = new \RedisSentinel($options);
  236. } else {
  237. $extra = $passAuth ? [$params['auth']] : [];
  238. $sentinel = @new $sentinelClass($host, $port, $params['timeout'], (string) $params['persistent_id'], $params['retry_interval'], $params['read_timeout'], ...$extra);
  239. }
  240. if ($address = @$sentinel->getMasterAddrByName($params['sentinel'])) {
  241. [$host, $port] = $address;
  242. }
  243. } catch (\RedisException|\Relay\Exception $redisException) {
  244. }
  245. } while (++$hostIndex < \count($hosts) && !$address);
  246. if (isset($params['sentinel']) && !$address) {
  247. throw new InvalidArgumentException(\sprintf('Failed to retrieve master information from sentinel "%s".', $params['sentinel']), previous: $redisException ?? null);
  248. }
  249. try {
  250. $extra = [
  251. 'stream' => self::filterSslOptions($params['ssl'] ?? []) ?: null,
  252. ];
  253. if (null !== $params['auth']) {
  254. $extra['auth'] = $params['auth'];
  255. }
  256. @$redis->{$connect}($host, $port, (float) $params['timeout'], (string) $params['persistent_id'], $params['retry_interval'], $params['read_timeout'], ...\defined('Redis::SCAN_PREFIX') || !$isRedisExt ? [$extra] : []);
  257. set_error_handler(function ($type, $msg) use (&$error) { $error = $msg; });
  258. try {
  259. $isConnected = $redis->isConnected();
  260. } finally {
  261. restore_error_handler();
  262. }
  263. if (!$isConnected) {
  264. $error = preg_match('/^Redis::p?connect\(\): (.*)/', $error ?? $redis->getLastError() ?? '', $error) ? \sprintf(' (%s)', $error[1]) : '';
  265. throw new InvalidArgumentException('Redis connection failed: '.$error.'.');
  266. }
  267. if (0 < $params['tcp_keepalive'] && (!$isRedisExt || \defined('Redis::OPT_TCP_KEEPALIVE'))) {
  268. $redis->setOption($isRedisExt ? \Redis::OPT_TCP_KEEPALIVE : Relay::OPT_TCP_KEEPALIVE, $params['tcp_keepalive']);
  269. }
  270. if ((!\defined('Redis::SCAN_PREFIX') && null !== $auth && $isRedisExt && !$redis->auth($auth)) || !$redis->select($params['dbindex'])) {
  271. $e = preg_replace('/^ERR /', '', $redis->getLastError());
  272. throw new InvalidArgumentException('Redis connection failed: '.$e.'.');
  273. }
  274. } catch (\RedisException|\Relay\Exception $e) {
  275. throw new InvalidArgumentException('Redis connection failed: '.$e->getMessage());
  276. }
  277. return $redis;
  278. };
  279. if ($params['lazy']) {
  280. $redis = $isRedisExt ? RedisProxy::createLazyProxy($initializer) : RelayProxy::createLazyProxy($initializer);
  281. } else {
  282. $redis = $initializer();
  283. }
  284. } elseif (is_a($class, \RedisArray::class, true)) {
  285. foreach ($hosts as $i => $host) {
  286. $hosts[$i] = match ($host['scheme']) {
  287. 'tcp' => $host['host'].':'.$host['port'],
  288. 'tls' => 'tls://'.$host['host'].':'.$host['port'],
  289. default => $host['path'],
  290. };
  291. }
  292. $params['lazy_connect'] = $params['lazy'] ?? true;
  293. $params['connect_timeout'] = $params['timeout'];
  294. try {
  295. $redis = new $class($hosts, $params);
  296. } catch (\RedisClusterException $e) {
  297. throw new InvalidArgumentException('Redis connection failed: '.$e->getMessage());
  298. }
  299. if (0 < $params['tcp_keepalive'] && (!$isRedisExt || \defined('Redis::OPT_TCP_KEEPALIVE'))) {
  300. $redis->setOption($isRedisExt ? \Redis::OPT_TCP_KEEPALIVE : Relay::OPT_TCP_KEEPALIVE, $params['tcp_keepalive']);
  301. }
  302. } elseif (is_a($class, RelayCluster::class, true)) {
  303. if (version_compare(phpversion('relay'), '0.10.0', '<')) {
  304. throw new InvalidArgumentException('Using RelayCluster is supported from ext-relay 0.10.0 or higher.');
  305. }
  306. $initializer = static function () use ($class, $params, $hosts) {
  307. foreach ($hosts as $i => $host) {
  308. $hosts[$i] = match ($host['scheme']) {
  309. 'tcp' => $host['host'].':'.$host['port'],
  310. 'tls' => 'tls://'.$host['host'].':'.$host['port'],
  311. default => $host['path'],
  312. };
  313. }
  314. try {
  315. $context = $params['cluster_relay_context'];
  316. $context['stream'] = self::filterSslOptions($params['ssl'] ?? []) ?: null;
  317. foreach ($context as $name => $value) {
  318. match ($name) {
  319. 'use-cache', 'client-tracking', 'throw-on-error', 'client-invalidations', 'reply-literal', 'persistent',
  320. => $context[$name] = filter_var($value, \FILTER_VALIDATE_BOOLEAN),
  321. 'max-retries', 'serializer', 'compression', 'compression-level',
  322. => $context[$name] = filter_var($value, \FILTER_VALIDATE_INT),
  323. default => null,
  324. };
  325. }
  326. $relayCluster = new $class(
  327. name: null,
  328. seeds: $hosts,
  329. connect_timeout: $params['timeout'],
  330. command_timeout: $params['cluster_command_timeout'],
  331. persistent: $params['persistent'],
  332. auth: $params['auth'] ?? null,
  333. context: $context,
  334. );
  335. } catch (\Relay\Exception $e) {
  336. throw new InvalidArgumentException('Relay cluster connection failed: '.$e->getMessage());
  337. }
  338. if (0 < $params['tcp_keepalive']) {
  339. $relayCluster->setOption(Relay::OPT_TCP_KEEPALIVE, $params['tcp_keepalive']);
  340. }
  341. if (0 < $params['read_timeout']) {
  342. $relayCluster->setOption(Relay::OPT_READ_TIMEOUT, $params['read_timeout']);
  343. }
  344. return $relayCluster;
  345. };
  346. $redis = $params['lazy'] ? RelayClusterProxy::createLazyProxy($initializer) : $initializer();
  347. } elseif (is_a($class, \RedisCluster::class, true)) {
  348. $initializer = static function () use ($isRedisExt, $class, $params, $hosts) {
  349. foreach ($hosts as $i => $host) {
  350. $hosts[$i] = match ($host['scheme']) {
  351. 'tcp' => $host['host'].':'.$host['port'],
  352. 'tls' => 'tls://'.$host['host'].':'.$host['port'],
  353. default => $host['path'],
  354. };
  355. }
  356. try {
  357. $redis = new $class(null, $hosts, $params['timeout'], $params['read_timeout'], $params['persistent'], $params['auth'] ?? '', ...\defined('Redis::SCAN_PREFIX') ? [$params['ssl'] ?? null] : []);
  358. } catch (\RedisClusterException $e) {
  359. throw new InvalidArgumentException('Redis connection failed: '.$e->getMessage());
  360. }
  361. if (0 < $params['tcp_keepalive'] && (!$isRedisExt || \defined('Redis::OPT_TCP_KEEPALIVE'))) {
  362. $redis->setOption($isRedisExt ? \Redis::OPT_TCP_KEEPALIVE : Relay::OPT_TCP_KEEPALIVE, $params['tcp_keepalive']);
  363. }
  364. $redis->setOption(\RedisCluster::OPT_SLAVE_FAILOVER, match ($params['failover']) {
  365. 'error' => \RedisCluster::FAILOVER_ERROR,
  366. 'distribute' => \RedisCluster::FAILOVER_DISTRIBUTE,
  367. 'slaves' => \RedisCluster::FAILOVER_DISTRIBUTE_SLAVES,
  368. 'none' => \RedisCluster::FAILOVER_NONE,
  369. });
  370. return $redis;
  371. };
  372. $redis = $params['lazy'] ? RedisClusterProxy::createLazyProxy($initializer) : $initializer();
  373. } elseif (is_a($class, \Predis\ClientInterface::class, true)) {
  374. if ($params['cluster']) {
  375. $params['cluster'] = 'redis';
  376. } else {
  377. unset($params['cluster']);
  378. }
  379. if (isset($params['sentinel'])) {
  380. $params['replication'] = 'sentinel';
  381. $params['service'] = $params['sentinel'];
  382. }
  383. $params += ['parameters' => []];
  384. $params['parameters'] += [
  385. 'persistent' => $params['persistent'],
  386. 'timeout' => $params['timeout'],
  387. 'read_write_timeout' => $params['read_timeout'],
  388. 'tcp_nodelay' => true,
  389. ];
  390. if ($params['dbindex']) {
  391. $params['parameters']['database'] = $params['dbindex'];
  392. }
  393. if (\is_array($params['auth'])) {
  394. // ACL
  395. $params['parameters']['username'] = $params['auth'][0];
  396. $params['parameters']['password'] = $params['auth'][1];
  397. } elseif (null !== $params['auth']) {
  398. $params['parameters']['password'] = $params['auth'];
  399. }
  400. if (isset($params['ssl'])) {
  401. foreach ($hosts as $i => $host) {
  402. $hosts[$i]['ssl'] ??= $params['ssl'];
  403. }
  404. }
  405. if (1 === \count($hosts) && !isset($params['cluster']) & !isset($params['sentinel'])) {
  406. $hosts = $hosts[0];
  407. } elseif (\in_array($params['failover'], ['slaves', 'distribute'], true) && !isset($params['replication'])) {
  408. $params['replication'] = true;
  409. $hosts[0] += ['alias' => 'master'];
  410. }
  411. $params['exceptions'] = false;
  412. $redis = new $class($hosts, array_diff_key($params, array_diff_key(self::$defaultConnectionOptions, ['cluster' => null])));
  413. if (isset($params['sentinel'])) {
  414. $redis->getConnection()->setSentinelTimeout($params['timeout']);
  415. }
  416. } elseif (class_exists($class, false)) {
  417. throw new InvalidArgumentException(\sprintf('"%s" is not a subclass of "Redis", "RedisArray", "RedisCluster", "Relay\Relay" nor "Predis\ClientInterface".', $class));
  418. } else {
  419. throw new InvalidArgumentException(\sprintf('Class "%s" does not exist.', $class));
  420. }
  421. return $redis;
  422. }
  423. protected function doFetch(array $ids): iterable
  424. {
  425. if (!$ids) {
  426. return [];
  427. }
  428. $result = [];
  429. if (($this->redis instanceof \Predis\ClientInterface && ($this->redis->getConnection() instanceof ClusterInterface || $this->redis->getConnection() instanceof Predis2ClusterInterface)) || $this->redis instanceof RelayCluster) {
  430. $values = $this->pipeline(function () use ($ids) {
  431. foreach ($ids as $id) {
  432. yield 'get' => [$id];
  433. }
  434. });
  435. } else {
  436. $values = $this->redis->mget($ids);
  437. if (!\is_array($values) || \count($values) !== \count($ids)) {
  438. return [];
  439. }
  440. $values = array_combine($ids, $values);
  441. }
  442. foreach ($values as $id => $v) {
  443. if ($v) {
  444. $result[$id] = $this->marshaller->unmarshall($v);
  445. }
  446. }
  447. return $result;
  448. }
  449. protected function doHave(string $id): bool
  450. {
  451. return (bool) $this->redis->exists($id);
  452. }
  453. protected function doClear(string $namespace): bool
  454. {
  455. if ($this->redis instanceof \Predis\ClientInterface) {
  456. $prefix = $this->redis->getOptions()->prefix ? $this->redis->getOptions()->prefix->getPrefix() : '';
  457. $prefixLen = \strlen($prefix ?? '');
  458. }
  459. $cleared = true;
  460. if ($this->redis instanceof RelayCluster) {
  461. $prefix = Relay::SCAN_PREFIX & $this->redis->getOption(Relay::OPT_SCAN) ? '' : $this->redis->getOption(Relay::OPT_PREFIX);
  462. $prefixLen = \strlen($prefix);
  463. $pattern = $prefix.$namespace.'*';
  464. foreach ($this->redis->_masters() as $ipAndPort) {
  465. $address = implode(':', $ipAndPort);
  466. $cursor = null;
  467. do {
  468. $keys = $this->redis->scan($cursor, $address, $pattern, 1000);
  469. if (isset($keys[1]) && \is_array($keys[1])) {
  470. $cursor = $keys[0];
  471. $keys = $keys[1];
  472. }
  473. if ($keys) {
  474. if ($prefixLen) {
  475. foreach ($keys as $i => $key) {
  476. $keys[$i] = substr($key, $prefixLen);
  477. }
  478. }
  479. $this->doDelete($keys);
  480. }
  481. } while ($cursor);
  482. }
  483. return $cleared;
  484. }
  485. $hosts = $this->getHosts();
  486. $host = reset($hosts);
  487. if ($host instanceof \Predis\Client) {
  488. $connection = $host->getConnection();
  489. if ($connection instanceof ReplicationInterface) {
  490. $hosts = [$host->getClientFor('master')];
  491. } elseif ($connection instanceof Predis2ReplicationInterface) {
  492. $connection->switchToMaster();
  493. $hosts = [$host];
  494. }
  495. }
  496. foreach ($hosts as $host) {
  497. if (!isset($namespace[0])) {
  498. $cleared = $host->flushDb() && $cleared;
  499. continue;
  500. }
  501. $info = $host->info('Server');
  502. $info = !$info instanceof ErrorInterface ? $info['Server'] ?? $info : ['redis_version' => '2.0'];
  503. if ($host instanceof Relay) {
  504. $prefix = Relay::SCAN_PREFIX & $host->getOption(Relay::OPT_SCAN) ? '' : $host->getOption(Relay::OPT_PREFIX);
  505. $prefixLen = \strlen($host->getOption(Relay::OPT_PREFIX) ?? '');
  506. } elseif (!$host instanceof \Predis\ClientInterface) {
  507. $prefix = \defined('Redis::SCAN_PREFIX') && (\Redis::SCAN_PREFIX & $host->getOption(\Redis::OPT_SCAN)) ? '' : $host->getOption(\Redis::OPT_PREFIX);
  508. $prefixLen = \strlen($host->getOption(\Redis::OPT_PREFIX) ?? '');
  509. }
  510. $pattern = $prefix.$namespace.'*';
  511. if (!version_compare($info['redis_version'], '2.8', '>=')) {
  512. // As documented in Redis documentation (http://redis.io/commands/keys) using KEYS
  513. // can hang your server when it is executed against large databases (millions of items).
  514. // Whenever you hit this scale, you should really consider upgrading to Redis 2.8 or above.
  515. $unlink = version_compare($info['redis_version'], '4.0', '>=') ? 'UNLINK' : 'DEL';
  516. $args = $this->redis instanceof \Predis\ClientInterface ? [0, $pattern] : [[$pattern], 0];
  517. $cleared = $host->eval("local keys=redis.call('KEYS',ARGV[1]) for i=1,#keys,5000 do redis.call('$unlink',unpack(keys,i,math.min(i+4999,#keys))) end return 1", $args[0], $args[1]) && $cleared;
  518. continue;
  519. }
  520. $cursor = null;
  521. do {
  522. $keys = $host instanceof \Predis\ClientInterface ? $host->scan($cursor ?? 0, 'MATCH', $pattern, 'COUNT', 1000) : $host->scan($cursor, $pattern, 1000);
  523. if (isset($keys[1]) && \is_array($keys[1])) {
  524. $cursor = $keys[0];
  525. $keys = $keys[1];
  526. }
  527. if ($keys) {
  528. if ($prefixLen) {
  529. foreach ($keys as $i => $key) {
  530. $keys[$i] = substr($key, $prefixLen);
  531. }
  532. }
  533. $this->doDelete($keys);
  534. }
  535. } while ($cursor);
  536. }
  537. return $cleared;
  538. }
  539. protected function doDelete(array $ids): bool
  540. {
  541. if (!$ids) {
  542. return true;
  543. }
  544. if ($this->redis instanceof \Predis\ClientInterface && ($this->redis->getConnection() instanceof ClusterInterface || $this->redis->getConnection() instanceof Predis2ClusterInterface)) {
  545. static $del;
  546. $del ??= (class_exists(UNLINK::class) ? 'unlink' : 'del');
  547. $this->pipeline(function () use ($ids, $del) {
  548. foreach ($ids as $id) {
  549. yield $del => [$id];
  550. }
  551. })->rewind();
  552. } else {
  553. static $unlink = true;
  554. if ($unlink) {
  555. try {
  556. $unlink = false !== $this->redis->unlink($ids);
  557. } catch (\Throwable) {
  558. $unlink = false;
  559. }
  560. }
  561. if (!$unlink) {
  562. $this->redis->del($ids);
  563. }
  564. }
  565. return true;
  566. }
  567. protected function doSave(array $values, int $lifetime): array|bool
  568. {
  569. if (!$values = $this->marshaller->marshall($values, $failed)) {
  570. return $failed;
  571. }
  572. $results = $this->pipeline(function () use ($values, $lifetime) {
  573. foreach ($values as $id => $value) {
  574. if (0 >= $lifetime) {
  575. yield 'set' => [$id, $value];
  576. } else {
  577. yield 'setEx' => [$id, $lifetime, $value];
  578. }
  579. }
  580. });
  581. foreach ($results as $id => $result) {
  582. if (true !== $result && (!$result instanceof Status || Status::get('OK') !== $result)) {
  583. $failed[] = $id;
  584. }
  585. }
  586. return $failed;
  587. }
  588. private function pipeline(\Closure $generator, ?object $redis = null): \Generator
  589. {
  590. $ids = [];
  591. $redis ??= $this->redis;
  592. if ($redis instanceof \RedisCluster || $redis instanceof RelayCluster || ($redis instanceof \Predis\ClientInterface && ($redis->getConnection() instanceof RedisCluster || $redis->getConnection() instanceof Predis2RedisCluster))) {
  593. // phpredis & predis don't support pipelining with RedisCluster
  594. // \Relay\Cluster does not support multi with pipeline mode
  595. // see https://github.com/phpredis/phpredis/blob/develop/cluster.markdown#pipelining
  596. // see https://github.com/nrk/predis/issues/267#issuecomment-123781423
  597. $results = [];
  598. foreach ($generator() as $command => $args) {
  599. $results[] = $redis->{$command}(...$args);
  600. $ids[] = 'eval' === $command ? ($redis instanceof \Predis\ClientInterface ? $args[2] : $args[1][0]) : $args[0];
  601. }
  602. } elseif ($redis instanceof \Predis\ClientInterface) {
  603. $results = $redis->pipeline(static function ($redis) use ($generator, &$ids) {
  604. foreach ($generator() as $command => $args) {
  605. $redis->{$command}(...$args);
  606. $ids[] = 'eval' === $command ? $args[2] : $args[0];
  607. }
  608. });
  609. } elseif ($redis instanceof \RedisArray) {
  610. $connections = $results = [];
  611. foreach ($generator() as $command => $args) {
  612. $id = 'eval' === $command ? $args[1][0] : $args[0];
  613. if (!isset($connections[$h = $redis->_target($id)])) {
  614. $connections[$h] = [$redis->_instance($h), -1];
  615. $connections[$h][0]->multi(\Redis::PIPELINE);
  616. }
  617. $connections[$h][0]->{$command}(...$args);
  618. $results[] = [$h, ++$connections[$h][1]];
  619. $ids[] = $id;
  620. }
  621. foreach ($connections as $h => $c) {
  622. $connections[$h] = $c[0]->exec();
  623. }
  624. foreach ($results as $k => [$h, $c]) {
  625. $results[$k] = $connections[$h][$c];
  626. }
  627. } else {
  628. $redis->multi($redis instanceof Relay ? Relay::PIPELINE : \Redis::PIPELINE);
  629. foreach ($generator() as $command => $args) {
  630. $redis->{$command}(...$args);
  631. $ids[] = 'eval' === $command ? $args[1][0] : $args[0];
  632. }
  633. $results = $redis->exec();
  634. }
  635. if (!$redis instanceof \Predis\ClientInterface && 'eval' === $command && $redis->getLastError()) {
  636. $e = $redis instanceof Relay ? new \Relay\Exception($redis->getLastError()) : new \RedisException($redis->getLastError());
  637. $results = array_map(fn ($v) => false === $v ? $e : $v, (array) $results);
  638. }
  639. if (\is_bool($results)) {
  640. return;
  641. }
  642. foreach ($ids as $k => $id) {
  643. yield $id => $results[$k];
  644. }
  645. }
  646. private function getHosts(): array
  647. {
  648. $hosts = [$this->redis];
  649. if ($this->redis instanceof \Predis\ClientInterface) {
  650. $connection = $this->redis->getConnection();
  651. if (($connection instanceof ClusterInterface || $connection instanceof Predis2ClusterInterface) && $connection instanceof \Traversable) {
  652. $hosts = [];
  653. foreach ($connection as $c) {
  654. $hosts[] = new \Predis\Client($c);
  655. }
  656. }
  657. } elseif ($this->redis instanceof \RedisArray) {
  658. $hosts = [];
  659. foreach ($this->redis->_hosts() as $host) {
  660. $hosts[] = $this->redis->_instance($host);
  661. }
  662. } elseif ($this->redis instanceof \RedisCluster) {
  663. $hosts = [];
  664. foreach ($this->redis->_masters() as $host) {
  665. $hosts[] = new RedisClusterNodeProxy($host, $this->redis);
  666. }
  667. }
  668. return $hosts;
  669. }
  670. private static function filterSslOptions(array $options): array
  671. {
  672. foreach ($options as $name => $value) {
  673. match ($name) {
  674. 'allow_self_signed', 'capture_peer_cert', 'capture_peer_cert_chain', 'disable_compression', 'SNI_enabled', 'verify_peer', 'verify_peer_name',
  675. => $options[$name] = filter_var($value, \FILTER_VALIDATE_BOOLEAN),
  676. default => null,
  677. };
  678. }
  679. return $options;
  680. }
  681. }