mirror of
https://github.com/predis/predis.git
synced 2026-08-30 20:21:31 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5a1430e99c |
@@ -0,0 +1,175 @@
|
|||||||
|
<?php
|
||||||
|
|
||||||
|
/*
|
||||||
|
* This file is part of the Predis package.
|
||||||
|
*
|
||||||
|
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||||
|
*
|
||||||
|
* For the full copyright and license information, please view the LICENSE
|
||||||
|
* file that was distributed with this source code.
|
||||||
|
*/
|
||||||
|
|
||||||
|
namespace Predis\Replication;
|
||||||
|
|
||||||
|
use Predis\ClientContextInterface;
|
||||||
|
use Predis\ClientException;
|
||||||
|
use Predis\ClientInterface;
|
||||||
|
use Predis\Command\CommandInterface;
|
||||||
|
use Predis\Command\RawCommand;
|
||||||
|
use Predis\Connection\ConnectionInterface;
|
||||||
|
use Predis\Connection\Cluster\ClusterInterface;
|
||||||
|
use Predis\Connection\Replication\ReplicationInterface;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Abstraction for the WAIT command.
|
||||||
|
*
|
||||||
|
* This can be used with a client connected to a single node or configured with
|
||||||
|
* both cluster and replication backends.
|
||||||
|
*
|
||||||
|
* @see http://redis.io/commands/wait
|
||||||
|
*
|
||||||
|
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||||
|
*/
|
||||||
|
class WaitContext
|
||||||
|
{
|
||||||
|
/**
|
||||||
|
* @var ClientInterface
|
||||||
|
*/
|
||||||
|
private $client;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @var CommandInterface[]
|
||||||
|
*/
|
||||||
|
private $commands = array();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param ClientInterface $client Client instance.
|
||||||
|
*/
|
||||||
|
public function __construct(ClientInterface $client)
|
||||||
|
{
|
||||||
|
$this->client = $client;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@inheritdoc}
|
||||||
|
*/
|
||||||
|
public function __call($commandID, $arguments)
|
||||||
|
{
|
||||||
|
$command = $this->client->createCommand($commandID, $arguments);
|
||||||
|
$response = $this->executeCommand($command);
|
||||||
|
|
||||||
|
return $response;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@inheritdoc}
|
||||||
|
*/
|
||||||
|
public function executeCommand(CommandInterface $command)
|
||||||
|
{
|
||||||
|
$response = $this->client->executeCommand($command);
|
||||||
|
$this->commands[] = $command;
|
||||||
|
|
||||||
|
return $response;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Executes the WAIT command against a connection.
|
||||||
|
*
|
||||||
|
* @param ConnectionInterface $connection [description]
|
||||||
|
* @param int $numslaves Minimum number of slaves for acknowledgment.
|
||||||
|
* @param int $timeout Timeout in milliseconds for acknowledgment.
|
||||||
|
*
|
||||||
|
* @return int
|
||||||
|
*/
|
||||||
|
private function executeWaitCommand(ConnectionInterface $connection, $numslaves, $timeout)
|
||||||
|
{
|
||||||
|
$response = $connection->executeCommand(
|
||||||
|
RawCommand::create('WAIT', $numslaves, $timeout)
|
||||||
|
);
|
||||||
|
|
||||||
|
return $response;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the connection for WAIT from a replication backend.
|
||||||
|
*
|
||||||
|
* @param ReplicationInterface $replication Replication connection backend.
|
||||||
|
*
|
||||||
|
* @return ConnectionInterface
|
||||||
|
*/
|
||||||
|
private function getConnectionFromReplication(ReplicationInterface $replication)
|
||||||
|
{
|
||||||
|
return $replication->getCurrent();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the connection for WAIT from a cluster backend.
|
||||||
|
*
|
||||||
|
* @param ClusterInterface $cluster Cluster connection backend.
|
||||||
|
*
|
||||||
|
* @return ConnectionInterface
|
||||||
|
*/
|
||||||
|
private function getConnectionFromCluster(ClusterInterface $cluster)
|
||||||
|
{
|
||||||
|
$command = reset($this->commands);
|
||||||
|
$slot = $command->getSlot();
|
||||||
|
|
||||||
|
foreach ($this->commands as $command) {
|
||||||
|
if ($slot !== $command->getSlot()) {
|
||||||
|
throw new ClientException('Cross-slot operations are not allowed');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO: we should actually fetch the connection by slot but this is not
|
||||||
|
// currently supported by all our cluster backends.
|
||||||
|
$connection = $cluster->getConnection($command);
|
||||||
|
|
||||||
|
return $connection;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Executes the WAIT command and returns the status of acknowledgment.
|
||||||
|
*
|
||||||
|
* When the client is operating in replication mode WAIT is executed against
|
||||||
|
* the connection currently in use by the underlying backend. On the other
|
||||||
|
* hand when it is operating in cluster mode WAIT is executed against only
|
||||||
|
* one connection as cross-slot operations are not allowed.
|
||||||
|
*
|
||||||
|
* @param int $numslaves Minimum number of slaves for acknowledgment.
|
||||||
|
* @param int $timeout Timeout in milliseconds for acknowledgment.
|
||||||
|
* @param int &$slaves Set with the number of slaves that acknowledged the writes.
|
||||||
|
*
|
||||||
|
* @return bool
|
||||||
|
*/
|
||||||
|
public function wait($numslaves, $timeout, &$slaves = 0)
|
||||||
|
{
|
||||||
|
if (!$this->commands) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
$connection = $this->client->getConnection();
|
||||||
|
|
||||||
|
if ($connection instanceof ReplicationInterface) {
|
||||||
|
$connection = $this->getConnectionFromReplication($connection);
|
||||||
|
} elseif ($connection instanceof ClusterInterface) {
|
||||||
|
$connection = $this->getConnectionFromCluster($connection);
|
||||||
|
}
|
||||||
|
|
||||||
|
$this->commands = array();
|
||||||
|
|
||||||
|
$slaves = $this->executeWaitCommand($connection, $numslaves, $timeout);
|
||||||
|
$acknowledged = $slaves >= $numslaves;
|
||||||
|
|
||||||
|
return $acknowledged;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the underlying client instance.
|
||||||
|
*
|
||||||
|
* @return ClientInterface
|
||||||
|
*/
|
||||||
|
public function getClient()
|
||||||
|
{
|
||||||
|
return $this->client;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user