multi(); } /** * Hijack `multi()` calls to allow command logging. * * @param ?int $type * @return object */ public function multi(?int $type = null) { return Transaction::multi($this); } /** * Yields all keys matching the given pattern. * * @param string|null $pattern * @param int $limit * @return \Generator> */ public function listKeys(?string $pattern = null, int $limit = 500): Generator { foreach ($this->client->_masters() as $primary) { $iterator = null; do { $keys = $this->client->scan($iterator, $primary, $pattern, $limit); if (! empty($keys)) { yield $keys; } } while ($iterator > 0); } } /** * Pings first primary node. * * To ping a specific node, pass name of key as a string, or a hostname and port as array. * * @param string|array $parameter * @return bool */ public function ping($parameter = null) { if (\is_null($parameter)) { $primaries = $this->client->_masters(); $parameter = \reset($primaries); } return $this->command('ping', [$parameter]); } /** * Fetches information from the first primary node. * * To fetch information from a specific node, pass the key or a host:port as array. * * @param string|null $section * @param string|array $key * @return bool */ public function info($section = null, $key = null) { $sections = [ 'server', 'clients', 'persistence', 'threads', 'cpu', 'memory', 'sentinel', 'cluster', 'replication', 'modules', 'keyspace', 'stats', 'errorstats', 'latencystats', 'commandstats', 'all', 'default', 'everything', ]; $validSection = \is_string($section) && \in_array(\strtolower($section), $sections); // support old `info($key_or_address)` signature if (\is_null($key) && (\is_array($section) || ! $validSection)) { $key = $section; $section = null; } if (\is_null($key)) { $primaries = $this->client->_masters(); $key = \reset($primaries); } return $this->command('info', array_filter([$key, $section])); } /** * Call `EVAL` script one key at a time and on all primaries when needed. * * @param string $script * @param array $args * @param int $keys * @return mixed */ public function eval(string $script, array $args = [], int $keys = 0) { if ($keys === 0) { // Will go to random primary return $this->command('eval', [$script, $args, 0]); } if ($keys === 1) { if (strpos($args[0], '{') === false) { // Must be run on all primaries return $this->evalWithoutHashTag($script, $args, 1); } // Will be called on the primary matching the hash-tag of the key return $this->command('eval', [$script, $args, 1]); } $results = []; foreach (array_slice($args, 0, $keys) as $key) { // Call this method recursively for each key $results[$key] = $this->eval($script, array_merge([$key], array_slice($args, $keys)), 1); } return $results; } /** * Call `EVAL` script on all primary nodes. * * @param string $script * @param array $args * @param int $keys * @return mixed */ protected function evalWithoutHashTag(string $script, array $args = [], int $keys = 0) { $results = []; $primaries = $this->client->_masters(); foreach ($primaries as $primary) { $key = $this->randomKey($primary); if ($key) { $results[] = $this->command('eval', [$script, array_merge([$key], $args, ['use-argv']), 1]); } } return $results; } /** * Returns Redis cluster nodes. * * @return array> */ public function nodes() { $nodes = $this->rawCommand( $this->client->_masters()[0], 'CLUSTER', 'NODES' ); $nodes = array_filter(explode("\n", $nodes)); return array_map(function ($node) { $fields = explode(' ', $node); return [ 'address' => strstr($fields[1], '@', true), 'flags' => str_replace(['myself,', 'myself'], '', $fields[2]), ]; }, $nodes); } /** * Flush all nodes on the Redis cluster. * * @param bool|null $async * @return true */ public function flushdb($async = null) { $useAsync = $async ?? $this->config->async_flush; foreach ($this->client->_masters() as $primary) { $useAsync ? $this->rawCommand($primary, 'flushdb', 'async') : $this->command('flushdb', [$primary]); } return true; } /** * Handles `CONFIG` calls for cluster connections. * * @param string $command * @param string ...$args * @return mixed */ public function config($command, ...$args) { $command = strtolower($command); if ($command == 'get') { $primaries = $this->client->_masters(); array_unshift($args, reset($primaries), $command); $config = $this->command('config', $args); if (is_array($config) && array_keys($config) === range(0, count($config) - 1)) { return array_column(array_chunk($config, 2), 1, 0); } return $config; } if ($command == 'resetstat') { return array_reduce($this->client->_masters(), function ($carry, $primary) { return $this->command('config', [$primary, 'resetstat']) && $carry; }, true); } throw new ConnectionException("config|{$command} not implemented for cluster connections."); } /** * Handles `SLOWLOG` calls for cluster connections. * * @param string $command * @param string ...$args * @return mixed */ public function slowlog($command, ...$args) { $command = strtolower($command); if ($command == 'get') { $log = []; foreach ($this->client->_masters() as $primary) { $result = $this->client->rawCommand($primary, 'slowlog', $command, ...$args); if ($result === false) { return false; } foreach ($result as $item) { $key = implode('|', array_slice($item[3], 0, 2)); if (! array_key_exists($key, $log) || $log[$key][2] < $item[2]) { $log[$key] = $item; } } } return array_values($log); } if ($command == 'reset') { return array_reduce($this->client->_masters(), function ($carry, $primary) { return $this->client->rawCommand($primary, 'slowlog', 'reset') && $carry; }, true); } throw new ConnectionException("slowlog|{$command} not implemented for cluster connections."); } }