Make sure lazy client follows exactly Client semantics

Author: clue

README.md

@@ -47,23 +47,22 @@ local Redis server and send some requests:
 $loop = React\EventLoop\Factory::create();
 $factory = new Factory($loop);
 
-$factory->createClient('localhost')->then(function (Client $client) use ($loop) {
-    $client->set('greeting', 'Hello world');
-    $client->append('greeting', '!');
-    
-    $client->get('greeting')->then(function ($greeting) {
-        // Hello world!
-        echo $greeting . PHP_EOL;
-    });
-    
-    $client->incr('invocation')->then(function ($n) {
-        echo 'This is invocation #' . $n . PHP_EOL;
-    });
-    
-    // end connection once all pending requests have been resolved
-    $client->end();
+$client = $factory->createLazyClient('localhost');
+$client->set('greeting', 'Hello world');
+$client->append('greeting', '!');
+
+$client->get('greeting')->then(function ($greeting) {
+    // Hello world!
+    echo $greeting . PHP_EOL;
+});
+
+$client->incr('invocation')->then(function ($n) {
+    echo 'This is invocation #' . $n . PHP_EOL;
 });
 
+// end connection once all pending requests have been resolved
+$client->end();
+
 $loop->run();
 ```
 
@@ -101,7 +100,7 @@ $factory = new Factory($loop, $connector);
 
 #### createClient()
 
-The `createClient($redisUri): PromiseInterface<Client,Exception>` method can be used to
+The `createClient(string $redisUri): PromiseInterface<Client,Exception>` method can be used to
 create a new [`Client`](#client).
 
 It helps with establishing a plain TCP/IP or secure TLS connection to Redis
@@ -198,9 +197,103 @@ $factory->createClient('localhost?timeout=0.5');
 
 #### createLazyClient()
 
-The `createLazyClient($redisUri)` method can be used to create a new [`Client`](#client) which lazily 
-creates and connects to the configured redis server on the first command. Internally it will use `createClient()` 
-when the first command comes in, queues all commands while connecting, and pass on all commands directly when connected.
+The `createLazyClient(string $redisUri): Client` method can be used to
+create a new [`Client`](#client).
+
+It helps with establishing a plain TCP/IP or secure TLS connection to Redis
+and optionally authenticating (AUTH) and selecting the right database (SELECT).
+
+```php
+$client = $factory->createLazyClient('redis://localhost:6379');
+
+$client->incr('hello');
+$client->end();
+```
+
+This method immediately returns a "virtual" connection implementing the
+[`Client`](#client) that can be used to interface with your Redis database.
+Internally, it lazily creates the underlying database connection (which may
+take some time) only once the first request is invoked on this instance and
+will queue all outstanding requests until the underlying connection is ready.
+
+From a consumer side this means that you can start sending commands to the
+database right away while the actual connection may still be outstanding.
+It will ensure that all commands will be executed in the order they are
+enqueued once the connection is ready. If the database connection fails,
+it will emit an `error` event, reject all outstanding commands and `close`
+the connection as described in the `Client`. In other words, it behaves just
+like a real connection and frees you from having to deal with its async
+resolution.
+
+Note that creating the underlying connection will be deferred until the
+first request is invoked. Accordingly, any eventual connection issues
+will be detected once this instance is first used. Similarly, calling
+`end()` on this instance before invoking any requests will succeed
+immediately and will not wait for an actual underlying connection.
+
+Depending on your particular use case, you may prefer this method or the
+underlying `createClient()` which resolves with a promise. For many
+simple use cases it may be easier to create a lazy connection.
+
+The `$redisUri` can be given in the
+[standard](https://www.iana.org/assignments/uri-schemes/prov/redis) form
+`[redis[s]://][:auth@]host[:port][/db]`.
+You can omit the URI scheme and port if you're connecting to the default port 6379:
+
+```php
+// both are equivalent due to defaults being applied
+$factory->createLazyClient('localhost');
+$factory->createLazyClient('redis://localhost:6379');
+```
+
+Redis supports password-based authentication (`AUTH` command). Note that Redis'
+authentication mechanism does not employ a username, so you can pass the
+password `h@llo` URL-encoded (percent-encoded) as part of the URI like this:
+
+```php
+// all forms are equivalent
+$factory->createLazyClient('redis://:h%40llo@localhost');
+$factory->createLazyClient('redis://ignored:h%40llo@localhost');
+$factory->createLazyClient('redis://localhost?password=h%40llo');
+```
+
+You can optionally include a path that will be used to select (SELECT command) the right database:
+
+```php
+// both forms are equivalent
+$factory->createLazyClient('redis://localhost/2');
+$factory->createLazyClient('redis://localhost?db=2');
+```
+
+You can use the [standard](https://www.iana.org/assignments/uri-schemes/prov/rediss)
+`rediss://` URI scheme if you're using a secure TLS proxy in front of Redis:
+
+```php
+$factory->createLazyClient('rediss://redis.example.com:6340');
+```
+
+You can use the `redis+unix://` URI scheme if your Redis instance is listening
+on a Unix domain socket (UDS) path:
+
+```php
+$factory->createLazyClient('redis+unix:///tmp/redis.sock');
+
+// the URI MAY contain `password` and `db` query parameters as seen above
+$factory->createLazyClient('redis+unix:///tmp/redis.sock?password=secret&db=2');
+
+// the URI MAY contain authentication details as userinfo as seen above
+// should be used with care, also note that database can not be passed as path
+$factory->createLazyClient('redis+unix://:secret@/tmp/redis.sock');
+```
+
+This method respects PHP's `default_socket_timeout` setting (default 60s)
+as a timeout for establishing the underlying connection and waiting for
+successful authentication. You can explicitly pass a custom timeout value
+in seconds (or use a negative number to not apply a timeout) like this:
+
+```php
+$factory->createLazyClient('localhost?timeout=0.5');
+```
 
 ### Client
 

examples/cli.php

@@ -11,10 +11,7 @@
 
 echo '# connecting to redis...' . PHP_EOL;
 
-/** @var Client $client */
-$client = $factory->createLazyClient('localhost');
-
-try {
+$factory->createClient('localhost')->then(function (Client $client) use ($loop) {
     echo '# connected! Entering interactive mode, hit CTRL-D to quit' . PHP_EOL;
 
     $loop->addReadStream(STDIN, function () use ($client, $loop) {
@@ -51,10 +48,10 @@
 
         $loop->removeReadStream(STDIN);
     });
-} catch (Exception $error) {
+}, function (Exception $error) {
     echo 'CONNECTION ERROR: ' . $error->getMessage() . PHP_EOL;
     exit(1);
-};
+});
 
 
 $loop->run();

examples/incr.php

@@ -1,14 +1,12 @@
 <?php
 
-use Clue\React\Redis\Client;
 use Clue\React\Redis\Factory;
 
 require __DIR__ . '/../vendor/autoload.php';
 
 $loop = React\EventLoop\Factory::create();
 $factory = new Factory($loop);
 
-/** @var Client $client */
 $client = $factory->createLazyClient('localhost');
 $client->incr('test');
 

examples/publish.php

@@ -1,6 +1,5 @@
 <?php
 
-use Clue\React\Redis\Client;
 use Clue\React\Redis\Factory;
 
 require __DIR__ . '/../vendor/autoload.php';
@@ -11,7 +10,6 @@
 $channel = isset($argv[1]) ? $argv[1] : 'channel';
 $message = isset($argv[2]) ? $argv[2] : 'message';
 
-/** @var Client $client */
 $client = $factory->createLazyClient('localhost');
 $client->publish($channel, $message)->then(function ($received) {
     echo 'successfully published. Received by ' . $received . PHP_EOL;

examples/subscribe.php

@@ -1,6 +1,5 @@
 <?php
 
-use Clue\React\Redis\Client;
 use Clue\React\Redis\Factory;
 
 require __DIR__ . '/../vendor/autoload.php';
@@ -10,7 +9,6 @@
 
 $channel = isset($argv[1]) ? $argv[1] : 'channel';
 
-/** @var Client $client */
 $client = $factory->createLazyClient('localhost');
 $client->subscribe($channel)->then(function () {
     echo 'Now subscribed to channel ' . PHP_EOL;

src/Factory.php

@@ -38,16 +38,11 @@ public function __construct(LoopInterface $loop, ConnectorInterface $connector =
         $this->protocol = $protocol;
     }
 
-    public function createLazyClient($target)
-    {
-        return new LazyStreamingClient($target, $this);
-    }
-
     /**
-     * create redis client connected to address of given redis instance
+     * Create Redis client connected to address of given redis instance
      *
      * @param string $target Redis server URI to connect to
-     * @return \React\Promise\PromiseInterface resolves with Client or rejects with \Exception
+     * @return \React\Promise\PromiseInterface<Client> resolves with Client or rejects with \Exception
      */
     public function createClient($target)
     {
@@ -120,6 +115,17 @@ function ($error) use ($client) {
         });
     }
 
+    /**
+     * Create Redis client connected to address of given redis instance
+     *
+     * @param string $target
+     * @return Client
+     */
+    public function createLazyClient($target)
+    {
+        return new LazyClient($target, $this);
+    }
+
     /**
      * @param string $target
      * @return array with keys authority, auth and db

src/LazyStreamingClient.php renamed to src/LazyClient.php

@@ -3,35 +3,20 @@
 namespace Clue\React\Redis;
 
 use Evenement\EventEmitter;
-use Clue\Redis\Protocol\Parser\ParserInterface;
-use Clue\Redis\Protocol\Parser\ParserException;
-use Clue\Redis\Protocol\Serializer\SerializerInterface;
-use Clue\Redis\Protocol\Factory as ProtocolFactory;
-use React\Promise\FulfilledPromise;
-use React\Promise\Promise;
 use React\Promise\PromiseInterface;
 use React\Stream\Util;
-use UnderflowException;
-use RuntimeException;
-use InvalidArgumentException;
-use React\Promise\Deferred;
-use Clue\Redis\Protocol\Model\ErrorReply;
-use Clue\Redis\Protocol\Model\ModelInterface;
-use Clue\Redis\Protocol\Model\MultiBulkReply;
-use React\Stream\DuplexStreamInterface;
 
 /**
  * @internal
  */
-class LazyStreamingClient extends EventEmitter implements Client
+class LazyClient extends EventEmitter implements Client
 {
     private $target;
     /** @var Factory */
     private $factory;
     private $ending = false;
     private $closed = false;
-    public $promise = null;
-    public $client = null;
+    private $promise;
 
     /**
      * @param $target
@@ -50,21 +35,13 @@ private function client()
             return $this->promise;
         }
 
-        if ($this->client instanceof Client) {
-            return new FulfilledPromise($this->client());
-        }
-
         $self = $this;
         return $this->promise = $this->factory->createClient($this->target)->then(function (Client $client) use ($self) {
-            $self->client = $client;
-            $self->promise = null;
-
             Util::forwardEvents(
-                $self->client,
+                $client,
                 $self,
                 array(
                     'error',
-                    'close',
                     'message',
                     'subscribe',
                     'unsubscribe',
@@ -74,6 +51,8 @@ private function client()
                 )
             );
 
+            $client->on('close', array($self, 'close'));
+
             return $client;
         }, function (\Exception $e) use ($self) {
             // connection failed => emit error if connection is not already closed
@@ -83,14 +62,14 @@ private function client()
             $self->emit('error', array($e));
             $self->close();
 
-            return $e;
+            throw $e;
         });
     }
 
     public function __call($name, $args)
     {
-        if ($this->client instanceof Client) {
-            return \call_user_func_array(array($this->client, $name), $args);
+        if ($this->closed) {
+            return \React\Promise\reject(new \RuntimeException('Connection closed'));
         }
 
         return $this->client()->then(function (Client $client) use ($name, $args) {
@@ -100,23 +79,37 @@ public function __call($name, $args)
 
     public function end()
     {
-        if ($this->client instanceof Client) {
-            return $this->client->end();
+        if ($this->promise === null) {
+            $this->close();
+        }
+
+        if ($this->closed) {
+            return;
         }
 
         return $this->client()->then(function (Client $client) {
-            return $client->end();
+            $client->end();
         });
     }
 
     public function close()
     {
-        if ($this->client instanceof Client) {
-            return $this->client->close();
+        if ($this->closed) {
+            return;
         }
 
-        return $this->client()->then(function (Client $client) {
-            return $client->close();
-        });
+        $this->closed = true;
+
+        // either close active connection or cancel pending connection attempt
+        if ($this->promise !== null) {
+            $this->promise->then(function (Client $client) {
+                $client->close();
+            });
+            $this->promise->cancel();
+            $this->promise = null;
+        }
+
+        $this->emit('close');
+        $this->removeAllListeners();
     }
 }