geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-09-10 18:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rename classes in the Predis\PubSub namespace.

Browse Source
This commit is contained in:
Daniele Alessandri
2013-11-16 15:27:58 +01:00
parent 7d60e995f8
commit c3a58dffdf
7 changed files with 70 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/PubSubContext.php → examples/PubSubConsumer.php
+2 -2
View File
@@ -17,7 +17,7 @@ require 'SharedConfigurations.php';
// Create a client and disable r/w timeout on the socket
$client = new Predis\Client($single_server + array('read_write_timeout' => 0));
// Initialize a new pubsub context
// Initialize a new pubsub consumer.
$pubsub = $client->pubSubLoop();
// Subscribe to your channels
@@ -49,7 +49,7 @@ foreach ($pubsub as $message) {
}
}
// Always unset the pubsub context instance when you are done! The
// Always unset the pubsub consumer instance when you are done! The
// class destructor will take care of cleanups and prevent protocol
// desynchronizations between the client and the server.
unset($pubsub);
lib/Predis/Client.php
+5 -5
View File
@@ -23,7 +23,7 @@ use Predis\Connection\ConnectionParametersInterface;
use Predis\Monitor\MonitorContext;
use Predis\Pipeline\PipelineContext;
use Predis\Profile\ServerProfile;
use Predis\PubSub\PubSubContext;
use Predis\PubSub;
use Predis\Response;
use Predis\Transaction\MultiExecContext;
@@ -421,7 +421,7 @@ class Client implements ClientInterface
* inside the optionally provided callable object.
*
* @param mixed $arg,... Options for the context, or a callable, or both.
* @return PubSubExecContext|NULL
* @return PubSub\Consumer|NULL
*/
public function pubSubLoop(/* arguments */)
{
@@ -433,11 +433,11 @@ class Client implements ClientInterface
*
* @param array $options Options for the context.
* @param mixed $callable Optional callable used to execute the context.
* @return PubSubContext|NULL
* @return PubSub\Consumer|NULL
*/
protected function createPubSub(Array $options = null, $callable = null)
{
$pubsub = new PubSubContext($this, $options);
$pubsub = new PubSub\Consumer($this, $options);
if (!isset($callable)) {
return $pubsub;
@@ -445,7 +445,7 @@ class Client implements ClientInterface
foreach ($pubsub as $message) {
if (call_user_func($callable, $pubsub, $message) === false) {
$pubsub->closeContext();
$pubsub->stop();
}
}
}
lib/Predis/PubSub/AbstractPubSubContext.php → lib/Predis/PubSub/AbstractConsumer.php
+9 -7
View File
@@ -11,12 +11,14 @@
namespace Predis\PubSub;
use Iterator;
/**
* Client-side abstraction of a Publish / Subscribe context.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
abstract class AbstractPubSubContext implements \Iterator
abstract class AbstractConsumer implements Iterator
{
const SUBSCRIBE = 'subscribe';
const UNSUBSCRIBE = 'unsubscribe';
@@ -37,7 +39,7 @@ abstract class AbstractPubSubContext implements \Iterator
*/
public function __destruct()
{
$this->closeContext(true);
$this->stop(true);
}
/**
@@ -98,16 +100,16 @@ abstract class AbstractPubSubContext implements \Iterator
* Optionally, the context can be forcefully closed by dropping the
* underlying connection.
*
* @param Boolean $force Forcefully close the context by closing the connection.
* @return Boolean Returns false if there are no pending messages.
* @param bool $drop Forcefully close the context by closing the connection.
* @return bool Returns false if there are no pending messages.
*/
public function closeContext($force = false)
public function stop($drop = false)
{
if (!$this->valid()) {
return false;
}
if ($force) {
if ($drop) {
$this->invalidate();
$this->disconnect();
} else {
@@ -119,7 +121,7 @@ abstract class AbstractPubSubContext implements \Iterator
}
}
return !$force;
return !$drop;
}
/**
lib/Predis/PubSub/PubSubContext.php → lib/Predis/PubSub/Consumer.php
+10 -5
View File
@@ -22,7 +22,7 @@ use Predis\Connection\AggregatedConnectionInterface;
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class PubSubContext extends AbstractPubSubContext
class Consumer extends AbstractConsumer
{
private $client;
private $options;
@@ -60,13 +60,17 @@ class PubSubContext extends AbstractPubSubContext
private function checkCapabilities(ClientInterface $client)
{
if ($client->getConnection() instanceof AggregatedConnectionInterface) {
throw new NotSupportedException('Cannot initialize a PUB/SUB context when using aggregated connections');
throw new NotSupportedException(
'Cannot initialize a PUB/SUB consumer 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');
throw new NotSupportedException(
'The current profile does not support PUB/SUB related commands'
);
}
}
@@ -132,8 +136,9 @@ class PubSubContext extends AbstractPubSubContext
);
default:
$message = "Received an unknown message type {$response[0]} inside of a pubsub context";
throw new ClientException($message);
throw new ClientException(
"Received an unknown message type {$response[0]} inside PUB/SUB"
);
}
}
}
lib/Predis/PubSub/DispatcherLoop.php
+13 -12
View File
@@ -11,6 +11,7 @@
namespace Predis\PubSub;
use InvalidArgumentException;
use Predis\ClientInterface;
/**
@@ -21,7 +22,7 @@ use Predis\ClientInterface;
*/
class DispatcherLoop
{
private $pubSubContext;
private $pubsub;
protected $callbacks;
protected $defaultCallback;
@@ -33,7 +34,7 @@ class DispatcherLoop
public function __construct(ClientInterface $client)
{
$this->callbacks = array();
$this->pubSubContext = $client->pubSubLoop();
$this->pubsub = $client->pubSubLoop();
}
/**
@@ -44,18 +45,18 @@ class DispatcherLoop
protected function validateCallback($callable)
{
if (!is_callable($callable)) {
throw new \InvalidArgumentException("A valid callable object must be provided");
throw new InvalidArgumentException("A valid callable object must be provided");
}
}
/**
* Returns the underlying Publish / Subscribe context.
*
* @return PubSubContext
* @return Consumer
*/
public function getPubSubContext()
public function getPubSubConsumer()
{
return $this->pubSubContext;
return $this->pubsub;
}
/**
@@ -99,7 +100,7 @@ class DispatcherLoop
$this->validateCallback($callback);
$this->callbacks[$callbackName] = $callback;
$this->pubSubContext->subscribe($channel);
$this->pubsub->subscribe($channel);
}
/**
@@ -113,7 +114,7 @@ class DispatcherLoop
if (isset($this->callbacks[$callbackName])) {
unset($this->callbacks[$callbackName]);
$this->pubSubContext->unsubscribe($channel);
$this->pubsub->unsubscribe($channel);
}
}
@@ -122,10 +123,10 @@ class DispatcherLoop
*/
public function run()
{
foreach ($this->pubSubContext as $message) {
foreach ($this->pubsub as $message) {
$kind = $message->kind;
if ($kind !== PubSubContext::MESSAGE && $kind !== PubSubContext::PMESSAGE) {
if ($kind !== Consumer::MESSAGE && $kind !== Consumer::PMESSAGE) {
if (isset($this->subscriptionCallback)) {
$callback = $this->subscriptionCallback;
call_user_func($callback, $message);
@@ -149,7 +150,7 @@ class DispatcherLoop
*/
public function stop()
{
$this->pubSubContext->closeContext();
$this->pubsub->stop();
}
/**
@@ -159,7 +160,7 @@ class DispatcherLoop
*/
protected function getPrefixKeys()
{
$options = $this->pubSubContext->getClient()->getOptions();
$options = $this->pubsub->getClient()->getOptions();
if (isset($options->prefix)) {
return $options->prefix->getPrefix();
tests/Predis/ClientTest.php
+4 -4
View File
@@ -635,24 +635,24 @@ class ClientTest extends StandardTestCase
/**
* @group disconnected
*/
public function testPubSubLoopWithoutArgumentsReturnsPubSubContext()
public function testPubSubLoopWithoutArgumentsReturnsPubSubConsumer()
{
$client = new Client();
$this->assertInstanceOf('Predis\PubSub\PubSubContext', $client->pubSubLoop());
$this->assertInstanceOf('Predis\PubSub\Consumer', $client->pubSubLoop());
}
/**
* @group disconnected
*/
public function testPubSubLoopWithArrayReturnsPubSubContextWithOptions()
public function testPubSubLoopWithArrayReturnsPubSubConsumerWithOptions()
{
$connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
$options = array('subscribe' => 'channel');
$client = new Client($connection);
$this->assertInstanceOf('Predis\PubSub\PubSubContext', $pubsub = $client->pubSubLoop($options));
$this->assertInstanceOf('Predis\PubSub\Consumer', $pubsub = $client->pubSubLoop($options));
$reflection = new \ReflectionProperty($pubsub, 'options');
$reflection->setAccessible(true);
tests/Predis/PubSub/PubSubContextTest.php → tests/Predis/PubSub/ConsumerTest.php
+27 -26
View File
@@ -15,18 +15,19 @@ use PHPUnit_Framework_TestCase as StandardTestCase;
use Predis\Client;
use Predis\Profile\ServerProfile;
use Predis\PubSub\Consumer as PubSubConsumer;
/**
* @group realm-pubsub
*/
class PubSubContextTest extends StandardTestCase
class ConsumerTest extends StandardTestCase
{
/**
* @group disconnected
* @expectedException Predis\NotSupportedException
* @expectedExceptionMessage The current profile does not support PUB/SUB related commands
*/
public function testPubSubContextRequirePubSubRelatedCommand()
public function testPubSubConsumerRequirePubSubRelatedCommand()
{
$profile = $this->getMock('Predis\Profile\ServerProfileInterface');
$profile->expects($this->any())
@@ -34,39 +35,39 @@ class PubSubContextTest extends StandardTestCase
->will($this->returnValue(false));
$client = new Client(null, array('profile' => $profile));
$pubsub = new PubSubContext($client);
$pubsub = new PubSubConsumer($client);
}
/**
* @group disconnected
* @expectedException Predis\NotSupportedException
* @expectedExceptionMessage Cannot initialize a PUB/SUB context when using aggregated connections
* @expectedExceptionMessage Cannot initialize a PUB/SUB consumer when using aggregated connections
*/
public function testPubSubContextDoesNotWorkOnClusters()
public function testPubSubConsumerDoesNotWorkOnClusters()
{
$cluster = $this->getMock('Predis\Connection\ClusterConnectionInterface');
$client = new Client($cluster);
$pubsub = new PubSubContext($client);
$pubsub = new PubSubConsumer($client);
}
/**
* @group disconnected
*/
public function testConstructorWithoutSubscriptionsDoesNotOpenContext()
public function testConstructorWithoutSubscriptionsDoesNotStartConsumer()
{
$connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
$client = $this->getMock('Predis\Client', array('executeCommand'), array($connection));
$client->expects($this->never())->method('executeCommand');
$pubsub = new PubSubContext($client);
$pubsub = new PubSubConsumer($client);
}
/**
* @group disconnected
*/
public function testConstructorWithSubscriptionsOpensContext()
public function testConstructorWithSubscriptionsStartsConsumer()
{
$profile = ServerProfile::get(REDIS_SERVER_VERSION);
@@ -85,30 +86,30 @@ class PubSubContextTest extends StandardTestCase
}));
$options = array('subscribe' => 'channel:foo', 'psubscribe' => 'channels:*');
$pubsub = new PubSubContext($client, $options);
$pubsub = new PubSubConsumer($client, $options);
}
/**
* @group disconnected
*/
public function testClosingContextWithTrueClosesConnection()
public function testStoppingConsumerWithTrueClosesConnection()
{
$connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
$client = $this->getMock('Predis\Client', array('disconnect'), array($connection));
$client->expects($this->exactly(1))->method('disconnect');
$pubsub = new PubSubContext($client, array('subscribe' => 'channel:foo'));
$pubsub = new PubSubConsumer($client, array('subscribe' => 'channel:foo'));
$connection->expects($this->never())->method('writeCommand');
$pubsub->closeContext(true);
$pubsub->stop(true);
}
/**
* @group disconnected
*/
public function testClosingContextWithFalseSendsUnsubscriptions()
public function testStoppingConsumerWithFalseSendsUnsubscriptions()
{
$profile = ServerProfile::get(REDIS_SERVER_VERSION);
$classUnsubscribe = $profile->getCommandClass('unsubscribe');
@@ -119,7 +120,7 @@ class PubSubContextTest extends StandardTestCase
$client = $this->getMock('Predis\Client', array('disconnect'), array($connection));
$options = array('subscribe' => 'channel:foo', 'psubscribe' => 'channels:*');
$pubsub = new PubSubContext($client, $options);
$pubsub = new PubSubConsumer($client, $options);
$connection->expects($this->exactly(2))
->method('writeCommand')
@@ -128,7 +129,7 @@ class PubSubContextTest extends StandardTestCase
$this->isInstanceOf($classPunsubscribe)
));
$pubsub->closeContext(false);
$pubsub->stop(false);
}
/**
@@ -139,7 +140,7 @@ class PubSubContextTest extends StandardTestCase
$connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
$client = $this->getMock('Predis\Client', array('disconnect'), array($connection));
$pubsub = new PubSubContext($client);
$pubsub = new PubSubConsumer($client);
$this->assertFalse($pubsub->valid());
$this->assertNull($pubsub->next());
@@ -156,7 +157,7 @@ class PubSubContextTest extends StandardTestCase
$connection->expects($this->once())->method('read')->will($this->returnValue($rawmessage));
$client = new Client($connection);
$pubsub = new PubSubContext($client, array('subscribe' => 'channel:foo'));
$pubsub = new PubSubConsumer($client, array('subscribe' => 'channel:foo'));
$message = $pubsub->current();
$this->assertSame('message', $message->kind);
@@ -175,7 +176,7 @@ class PubSubContextTest extends StandardTestCase
$connection->expects($this->once())->method('read')->will($this->returnValue($rawmessage));
$client = new Client($connection);
$pubsub = new PubSubContext($client, array('psubscribe' => 'channel:*'));
$pubsub = new PubSubConsumer($client, array('psubscribe' => 'channel:*'));
$message = $pubsub->current();
$this->assertSame('pmessage', $message->kind);
@@ -195,7 +196,7 @@ class PubSubContextTest extends StandardTestCase
$connection->expects($this->once())->method('read')->will($this->returnValue($rawmessage));
$client = new Client($connection);
$pubsub = new PubSubContext($client, array('subscribe' => 'channel:foo'));
$pubsub = new PubSubConsumer($client, array('subscribe' => 'channel:foo'));
$message = $pubsub->current();
$this->assertSame('subscribe', $message->kind);
@@ -214,7 +215,7 @@ class PubSubContextTest extends StandardTestCase
$connection->expects($this->once())->method('read')->will($this->returnValue($rawmessage));
$client = new Client($connection);
$pubsub = new PubSubContext($client, array('subscribe' => 'channel:foo'));
$pubsub = new PubSubConsumer($client, array('subscribe' => 'channel:foo'));
$message = $pubsub->current();
$this->assertSame('unsubscribe', $message->kind);
@@ -225,7 +226,7 @@ class PubSubContextTest extends StandardTestCase
/**
* @group disconnected
*/
public function testUnsubscriptionMessageWithZeroChannelCountInvalidatesContext()
public function testUnsubscriptionMessageWithZeroChannelCountInvalidatesConsumer()
{
$rawmessage = array('unsubscribe', 'channel:foo', 0);
@@ -233,7 +234,7 @@ class PubSubContextTest extends StandardTestCase
$connection->expects($this->once())->method('read')->will($this->returnValue($rawmessage));
$client = new Client($connection);
$pubsub = new PubSubContext($client, array('subscribe' => 'channel:foo'));
$pubsub = new PubSubConsumer($client, array('subscribe' => 'channel:foo'));
$this->assertTrue($pubsub->valid());
@@ -253,7 +254,7 @@ class PubSubContextTest extends StandardTestCase
$connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
$client = new Client($connection);
$pubsub = new PubSubContext($client);
$pubsub = new PubSubConsumer($client);
$this->assertSame($client, $pubsub->getClient());
}
@@ -284,7 +285,7 @@ class PubSubContextTest extends StandardTestCase
$consumer = new Client($parameters, $options);
$consumer->connect();
$pubsub = new PubSubContext($consumer);
$pubsub = new PubSubConsumer($consumer);
$pubsub->subscribe('channel:foo');
$producer->publish('channel:foo', 'message1');
@@ -297,7 +298,7 @@ class PubSubContextTest extends StandardTestCase
}
$messages[] = ($payload = $message->payload);
if ($payload === 'QUIT') {
$pubsub->closeContext();
$pubsub->stop();
}
}