php Predis-ClientInterface类(方法)实例源码

/**
  * {@inheritdoc}
  */
 public function __construct(ClientInterface $client)
 {
     if (!$client->getCommandFactory()->supportsCommands(array('multi', 'exec', 'discard'))) {
         throw new ClientException("'MULTI', 'EXEC' and 'DISCARD' are not supported by the current command factory.");
     }
     parent::__construct($client);
 }

/**
  * {@inheritdoc}
  */
 public function __construct(ClientInterface $client)
 {
     if (!$client->getProfile()->supportsCommands(array('multi', 'exec', 'discard'))) {
         throw new ClientException("The current profile does not support 'MULTI', 'EXEC' and 'DISCARD'.");
     }
     parent::__construct($client);
 }

/**
  * Returns a pipeline executor depending on the kind of the underlying
  * connection and the passed options.
  *
  * @param  ClientInterface           $client Client instance used by the context.
  * @return PipelineExecutorInterface
  */
 protected function createExecutor(ClientInterface $client)
 {
     $options = $client->getOptions();
     if (isset($options->exceptions)) {
         return new StandardExecutor($options->exceptions);
     }
     return new StandardExecutor();
 }

/**
  * Checks if the passed client instance satisfies the required conditions
  * needed to initialize a monitor consumer.
  *
  * @param ClientInterface $client Client instance used by the consumer.
  *
  * @throws NotSupportedException
  */
 private function assertClient(ClientInterface $client)
 {
     if ($client->getConnection() instanceof AggregateConnectionInterface) {
         throw new NotSupportedException('Cannot initialize a monitor consumer over aggregate connections.');
     }
     if ($client->getProfile()->supportsCommand('MONITOR') === false) {
         throw new NotSupportedException("The current profile does not support 'MONITOR'.");
     }
 }

/**
  * Checks if the passed client instance satisfies the required conditions
  * needed to initialize a monitor context.
  *
  * @param ClientInterface $client Client instance used by the context.
  */
 private function checkCapabilities(ClientInterface $client)
 {
     if ($client->getConnection() instanceof AggregatedConnectionInterface) {
         throw new NotSupportedException('Cannot initialize a monitor context when using aggregated connections');
     }
     if ($client->getProfile()->supportsCommand('monitor') === false) {
         throw new NotSupportedException('The current profile does not support the MONITOR command');
     }
 }

/**
  * Checks if the passed client instance satisfies the required conditions
  * needed to initialize a Publish / Subscribe context.
  *
  * @param ClientInterface $client Client instance used by the context.
  */
 private function checkCapabilities(ClientInterface $client)
 {
     if ($client->getConnection() instanceof AggregatedConnectionInterface) {
         throw new NotSupportedException('Cannot initialize a PUB/SUB context when using aggregated connections');
     }
     $commands = array('publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
     if ($client->getProfile()->supportsCommands($commands) === false) {
         throw new NotSupportedException('The current profile does not support PUB/SUB related commands');
     }
 }

/**
  * Checks if the client instance satisfies the required conditions needed to
  * initialize a PUB/SUB consumer.
  *
  * @param ClientInterface $client Client instance used by the consumer.
  *
  * @throws NotSupportedException
  */
 private function checkCapabilities(ClientInterface $client)
 {
     if ($client->getConnection() instanceof AggregateConnectionInterface) {
         throw new NotSupportedException('Cannot initialize a PUB/SUB consumer over aggregate connections.');
     }
     $commands = array('publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
     if ($client->getCommandFactory()->supportsCommands($commands) === false) {
         throw new NotSupportedException('PUB/SUB commands are not supported by the current command factory.');
     }
 }

/**
  * Configures the transaction using the provided options.
  *
  * @param ClientInterface $client  Underlying client instance.
  * @param array           $options Array of options for the transaction.
  * */
 protected function configure(ClientInterface $client, array $options)
 {
     if (isset($options['exceptions'])) {
         $this->exceptions = (bool) $options['exceptions'];
     } else {
         $this->exceptions = $client->getOptions()->exceptions;
     }
     if (isset($options['cas'])) {
         $this->modeCAS = (bool) $options['cas'];
     }
     if (isset($options['watch']) && ($keys = $options['watch'])) {
         $this->watchKeys = $keys;
     }
     if (isset($options['retry'])) {
         $this->attempts = (int) $options['retry'];
     }
 }

/**
  * @param ClientInterface $client Client instance used by the context.
  */
 public function __construct(ClientInterface $client)
 {
     $this->callbacks = array();
     $this->pubSubContext = $client->pubSub();
 }

/**
  * Ensures that the client supports the specified Redis command required to
  * fetch elements from the server to perform the iteration.
  *
  * @param ClientInterface $client    Client connected to Redis.
  * @param string          $commandID Command ID.
  *
  * @throws NotSupportedException
  */
 protected function requiredCommand(ClientInterface $client, $commandID)
 {
     if (!$client->getProfile()->supportsCommand($commandID)) {
         throw new NotSupportedException("The current profile does not support '{$commandID}'.");
     }
 }

/**
  * Checks if the passed client instance satisfies the required conditions
  * needed to initialize a transaction context.
  *
  * @param ClientInterface $client Client instance used by the context.
  */
 private function checkCapabilities(ClientInterface $client)
 {
     if ($client->getConnection() instanceof AggregatedConnectionInterface) {
         throw new NotSupportedException('Cannot initialize a MULTI/EXEC context when using aggregated connections');
     }
     $profile = $client->getProfile();
     if ($profile->supportsCommands(array('multi', 'exec', 'discard')) === false) {
         throw new NotSupportedException('The current profile does not support MULTI, EXEC and DISCARD');
     }
     $this->canWatch = $profile->supportsCommands(array('watch', 'unwatch'));
 }

/**
  * Broadcast the given event.
  *
  * @param  array $channels
  * @param  string $event
  * @param  array $payload
  * @return void
  */
 public function broadcast(array $channels, $event, array $payload = [])
 {
     $socket = isset($payload['socket']) ? $payload['socket'] : null;
     $payload = json_encode(['event' => $event, 'data' => $payload, 'socket' => $socket]);
     foreach ($this->formatChannels($channels) as $channel) {
         $this->redis->publish($channel, $payload);
     }
 }

/**
  * @inheritdoc
  */
 public function has(AggregateIdInterface $id, $version)
 {
     if (!$this->redis->hexists(static::KEY_NAMESPACE, (string) $id)) {
         return false;
     }
     $snapshot = $this->serializer->deserialize($this->redis->hget(static::KEY_NAMESPACE, (string) $id), 'array', 'json');
     return $snapshot['version'] === $version;
 }

/**
  * Set the cache for a token.
  *
  * @param $key
  * @param string $value
  * @param null $expiration
  */
 protected function cacheSet($key, $value, $expiration = null)
 {
     $this->client->set($key, $value);
     if (!empty($expiration)) {
         $this->ttl = $expiration;
     }
     $this->client->expire($key, $this->ttl);
 }

/**
  *
  * @throws QueueIsDraining If the Queue is Drainable
  *
  * @param string $message
  * @param array $options Message options (override Queue options):
  *
  * @return string Unique ID along the queue
  */
 protected function doEnqueue($message, $options = [])
 {
     // Score has form <priority>.<timestamp>
     $score = isset($options['score']) ? $options['score'] : 0;
     if ($priority = $this->getOption('priority', false, $options)) {
         $score += $priority;
     }
     $this->redis->zAdd($this->queueKey, $score, $message);
 }

/**
  * Releases the lock, if it has not been released already
  */
 public function release()
 {
     if ($this->released) {
         return;
     }
     // Only release the lock if it hasn't expired
     if ($this->expire >= time()) {
         $this->redis->del($this->key);
     }
     $this->released = true;
 }

private function deleteGlob($pattern)
 {
     $keys = $this->client->keys($pattern);
     $options = $this->client->getOptions();
     if (isset($options->prefix)) {
         $length = strlen($options->prefix->getPrefix());
         $keys = array_map(function ($key) use($length) {
             return substr($key, $length);
         }, $keys);
     }
     if (count($keys) === 0) {
         return;
     }
     $this->client->del($keys);
 }

/**
  * Remove an ID from an index
  * @param  string $id
  * @param  string $index
  * @return void
  */
 protected function removeIndexId(string $id, string $index)
 {
     if ($this->redis->scard($index) == 1) {
         return $this->redis->del($index);
     }
     return $this->redis->srem($index, $id);
 }

/**
  * @return array
  */
 public function getPosts() : array
 {
     if ($this->redisClient !== null) {
         $posts = $this->redisClient->get('posts');
         if ($posts) {
             return json_decode($posts, true);
         }
     }
     $blog = $this->getBlogInfo();
     $posts = $this->thumblrClient->getBlogPosts($blog['name'], ['type' => 'text'])->posts;
     $posts = json_encode($posts, JSON_PRETTY_PRINT);
     if ($this->redisClient !== null) {
         $this->redisClient->set('posts', $posts);
         $this->redisClient->expireat('posts', time() + 3600);
     }
     return json_decode($posts, true);
 }

/**
  * Release the lock if we possess it.
  *
  * Returns true if we released the lock, and false if it was already
  * released.
  *
  * @return bool
  */
 public function release()
 {
     if (!$this->locked()) {
         return false;
     }
     $this->redis->del($this->name);
     $this->state = self::UNLOCKED;
     return true;
 }