Throw ConnectionException when stream is not open (#76)

docs/getting-started.md

@@ -25,11 +25,9 @@ $connections = [
 $registry = new Registry($connections);
 
 $r = new Rethink($registry->getConnection('default_connection'));
-```
 
-The driver class `Rethink` has a default database defined in the connection options. However you can always switch database if needed.
-```php
-$r->use('demoDB-2');
+// Now you can connect to RethinkDB.
+$r->connection()->connect();
 ```
 
 The driver class `Rethink` has an API Interface that supports the ReQL domain-specific language (DSL).

readme.md

@@ -54,6 +54,9 @@ $connections = [
 $registry = new Registry($connections);
 
 $r = new Rethink($registry->getConnection('default_connection'));
+
+// Now you can connect to RethinkDB.
+$r->connection()->connect();
 ```
 
 The driver class `Rethink` has a default database defined in the connection options. However you can always switch database if needed.

src/Connection/Connection.php

@@ -23,30 +23,37 @@ class Connection implements ConnectionInterface, ConnectionCursorInterface
      * @var int[]
      */
     private $activeTokens;
+
     /**
      * @var string
      */
     private $dbName;
+
     /**
      * @var HandshakeInterface
      */
     private $handshake;
+
     /**
      * @var bool
      */
     private $noReply = false;
+
     /**
      * @var SerializerInterface
      */
     private $querySerializer;
+
     /**
      * @var SerializerInterface
      */
     private $responseSerializer;
+
     /**
      * @var StreamInterface
      */
     private $stream;
+
     /**
      * @var \Closure
      */
@@ -66,11 +73,6 @@ public function __construct(
         $this->responseSerializer = $responseSerializer;
     }
 
-    public function changes(MessageInterface $query): void
-    {
-        // TODO: Implement changes() method.
-    }
-
     /**
      * @throws \Exception
      */
@@ -241,6 +243,10 @@ public function use(string $name): void
      */
     public function writeQuery(int $token, MessageInterface $message): int
     {
+        if (!$this->stream) {
+            throw new ConnectionException('No open stream, please connect first');
+        }
+
         if ($this->dbName) {
             $message->setOptions((new QueryOptions())->setDb($this->dbName));
         }

src/Connection/ConnectionInterface.php

@@ -8,8 +8,6 @@
 
 interface ConnectionInterface
 {
-    public function changes(MessageInterface $query): void;
-
     public function close($noreplyWait = true): void;
 
     public function connect(): Connection;

test/unit/Connection/ConnectionTest.php

@@ -36,6 +36,15 @@ public function testConnectThrowsCorrectException(): void
         $this->connection->connect();
     }
 
+    /**
+     * @expectedException \TBolier\RethinkQL\Connection\ConnectionException
+     * @expectedExceptionMessage No open stream, please connect first
+     */
+    public function testQueryWithoutConnection(): void
+    {
+        $this->connection->writeQuery(1223456789, \Mockery::mock(MessageInterface::class));
+    }
+
     public function testExpr()
     {
         $this->connect();
@@ -115,22 +124,6 @@ public function testRunSequence()
         }
     }
 
-    /**
-     * @throws Exception
-     */
-    public function testChanges()
-    {
-        $message = \Mockery::mock(MessageInterface::class);
-
-        try {
-            $this->connection->changes($message);
-        } catch (\Exception $e) {
-            $this->fail($e->getMessage());
-        }
-
-        $this->assertTrue(true);
-    }
-
     /**
      * @throws Exception
      */