Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -231,9 +231,6 @@ Incoming events and errors will be forwarded to registered event handler callbac

```php
// global events:
$client->on('data', function (MessageInterface $message) {
// process an incoming message (raw message object)
});
$client->on('close', function () {
// the connection to Redis just closed
});
Expand Down
40 changes: 21 additions & 19 deletions examples/cli.php
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
<?php

use Clue\Redis\Protocol\Model\ErrorReply;
use Clue\React\Redis\Client;
use Clue\React\Redis\Factory;
use Clue\Redis\Protocol\Model\ModelInterface;
use React\Promise\PromiseInterface;

require __DIR__ . '/../vendor/autoload.php';

Expand All @@ -15,30 +14,33 @@
$factory->createClient()->then(function (Client $client) use ($loop) {
echo '# connected! Entering interactive mode, hit CTRL-D to quit' . PHP_EOL;

$client->on('data', function (ModelInterface $data) {
if ($data instanceof ErrorReply) {
echo '# error reply: ' . $data->getMessage() . PHP_EOL;
} else {
echo '# reply: ' . json_encode($data->getValueNative()) . PHP_EOL;
}
});

$loop->addReadStream(STDIN, function () use ($client, $loop) {
$line = fgets(STDIN);
if ($line === false || $line === '') {
echo '# CTRL-D -> Ending connection...' . PHP_EOL;
$client->end();
} else {
$line = rtrim($line);
$loop->removeReadStream(STDIN);
return $client->end();
}

if ($line === '') {
$line = rtrim($line);
if ($line === '') {
return;
}

$params = explode(' ', $line);
$method = array_shift($params);
$promise = call_user_func_array(array($client, $method), $params);

} else {
$params = explode(' ', $line);
$method = array_shift($params);
call_user_func_array(array($client, $method), $params);
}
// special method such as end() / close() called
if (!$promise instanceof PromiseInterface) {
return;
}

$promise->then(function ($data) {
echo '# reply: ' . json_encode($data) . PHP_EOL;
}, function ($e) {
echo '# error reply: ' . $e->getMessage() . PHP_EOL;
});
});

$client->on('close', function() use ($loop) {
Expand Down
2 changes: 0 additions & 2 deletions src/Client.php
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,10 @@

use Evenement\EventEmitterInterface;
use React\Promise\PromiseInterface;
use Clue\Redis\Protocol\Model\ModelInterface;

/**
* Simple interface for executing redis commands
*
* @event data(ModelInterface $messageModel)
* @event error(Exception $error)
* @event close()
*
Expand Down
2 changes: 0 additions & 2 deletions src/StreamingClient.php
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,6 @@ public function __call($name, $args)

public function handleMessage(ModelInterface $message)
{
$this->emit('data', array($message));

if ($this->monitoring && $this->isMonitorMessage($message)) {
$this->emit('monitor', array($message));
return;
Expand Down
13 changes: 0 additions & 13 deletions tests/StreamingClientTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -56,21 +56,8 @@ public function testReceiveParseErrorEmitsErrorEvent()
$this->stream->emit('data', array('message'));
}

public function testReceiveMessageEmitsEvent()
{
$this->client->on('data', $this->expectCallableOnce());

$this->parser->expects($this->once())->method('pushIncoming')->with($this->equalTo('message'))->will($this->returnValue(array(new IntegerReply(2))));
$this->stream->emit('data', array('message'));
}

public function testReceiveThrowMessageEmitsErrorEvent()
{
$this->client->on('data', $this->expectCallableOnce());
$this->client->on('data', function() {
throw new UnderflowException();
});

$this->client->on('error', $this->expectCallableOnce());

$this->parser->expects($this->once())->method('pushIncoming')->with($this->equalTo('message'))->will($this->returnValue(array(new IntegerReply(2))));
Expand Down