Auth capabilities and other improvements

Author: Anton Shabouta

CHANGELOG.md

@@ -7,7 +7,7 @@ We accept contributions via Pull Requests on [Github](https://github.com/phpinna
 
 ## Pull Requests
 
-- **[PSR-2 Coding Standard](https://github.com/php-fig/fig-standards/blob/master/accepted/PSR-2-coding-style-guide.md)** - Check the code style with ``$ composer check-style`` and fix it with ``$ composer fix-style``.
+- **Follow our coding standard.** - It mostly based on [PSR-2 Coding Standard](https://github.com/php-fig/fig-standards/blob/master/accepted/PSR-2-coding-style-guide.md).
 
 - **Add tests!** - Your patch won't be accepted if it does not have tests.
 
@@ -25,7 +25,7 @@ We accept contributions via Pull Requests on [Github](https://github.com/phpinna
 ## Running Tests
 
 ```bash
-$ composer test
+$ CASSIS_TEST_DSN=tcp://user:pass@localhost:9042 composer test
 ```
 
 

CONTRIBUTING.md

@@ -33,7 +33,7 @@ use PHPinnacle\Cassis\Statement;
 require __DIR__ . '/vendor/autoload.php';
 
 Loop::run(function () {
-    $cluster  = Cluster::build('tcp://admin:admin123@172.23.0.3');
+    $cluster = Cluster::build('tcp://localhost:9042');
 
     /** @var Session $session */
     $session = yield $cluster->connect('system');
@@ -60,14 +60,10 @@ CASSIS_EXAMPLE_DSN=tcp://user:pass@localhost:9042 php example/*
 Benchmarks were run as:
 
 ```bash
-
-CASSIS_BENCHMARK_DSN=tcp://user:pass@localhost:9042 php benchmark/simple.php N
+CASSIS_BENCHMARK_DSN=tcp://user:pass@localhost:9042 php benchmark/write.php N
+CASSIS_BENCHMARK_DSN=tcp://user:pass@localhost:9042 php benchmark/read.php N M
 ```
 
-## Change log
-
-Please see [CHANGELOG](CHANGELOG.md) for more information on what has changed recently.
-
 ## Testing
 
 ```bash

README.md

@@ -0,0 +1,91 @@
+<?php
+
+use Amp\Loop;
+use PHPinnacle\Cassis\Context;
+use PHPinnacle\Cassis\Session;
+use PHPinnacle\Cassis\Cluster;
+use PHPinnacle\Cassis\Value;
+use Ramsey\Uuid\Uuid;
+
+require_once __DIR__ . '/../vendor/autoload.php';
+
+Loop::run(function () use ($argv) {
+    if (!$dsn = \getenv('CASSIS_BENCHMARK_DSN')) {
+        echo 'No benchmark dsn! Please set CASSIS_BENCHMARK_DSN environment variable.', \PHP_EOL;
+
+        Loop::stop();
+    }
+
+    $cluster = Cluster::build(\getenv('CASSIS_BENCHMARK_DSN'));
+    /** @var Session $session */
+    $session = yield $cluster->connect();
+    $setup   = require __DIR__ . '/shared.php';
+
+    $watcher = Loop::onSignal(SIGTERM, function () use ($cluster) {
+        yield $cluster->disconnect();
+    });
+
+    try {
+        foreach ($setup as $query) {
+            yield $session->query($query);
+        }
+
+        $toWrite = $argv[1] ?? 10000;
+        $toRead  = $argv[2] ?? 1000;
+
+        $promises = [];
+
+        for ($i = 1; $i <= $toWrite; $i++) {
+            $author = new Value\UserDefined([
+                'id'      => $i,
+                'name'    => "User $i",
+                'enabled' => (bool) ($i % 2),
+            ]);
+
+            $arguments = [
+                'author'  => $author,
+                'post_id' => Uuid::uuid1(),
+                'text'    => random_string(500),
+                'date'    => Value\Timestamp::fromDateTime(random_date()),
+                'tags'    => Value\Collection::set(random_tags(\rand(1, 10), 5)),
+            ];
+
+            $fields = \implode(',', \array_keys($arguments));
+            $values = \implode(',', \array_fill(0, \count($arguments), '?'));
+
+            $promises[] = $session->query("INSERT INTO posts_by_user ($fields) VALUES ($values)", $arguments);
+        }
+
+        yield $promises;
+
+        echo \sprintf("Done %d inserts. Start reading...\n", $toWrite);
+
+        $time     = \microtime(true);
+        $context  = (new Context)->limit($toRead);
+        $total    = 0;
+
+        /** @var \PHPinnacle\Cassis\Result\Rows $result */
+        while ($result = yield $session->query("SELECT * FROM posts_by_user;", [], $context)) {
+            $count = \count($result);
+            $total = $total + $count;
+
+            echo \sprintf("Read %d rows.\n", $count);
+
+            if (!$cursor = $result->cursor()) {
+                break;
+            }
+
+            $context->offset($cursor);
+        };
+
+        echo \sprintf("Read %d rows in %f seconds.\n", $total, \microtime(true) - $time);
+    } catch (\Throwable $error) {
+        echo "Got error: {$error->getMessage()}.\n";
+    } finally {
+        yield $session->query("DROP KEYSPACE IF EXISTS blogs;");
+    }
+
+    yield $cluster->disconnect();
+
+    Loop::cancel($watcher);
+});

benchmark/read.php

@@ -15,7 +15,7 @@
         Loop::stop();
     }
 
-    $cluster = Cluster::build(\getenv('CASSIS_EXAMPLE_DSN'));
+    $cluster = Cluster::build(\getenv('CASSIS_BENCHMARK_DSN'));
     /** @var Session $session */
     $session = yield $cluster->connect();
     $setup   = require __DIR__ . '/shared.php';

benchmark/write.php

@@ -15,6 +15,7 @@
 use function Amp\call;
 use Amp\Promise;
 use Amp\Socket;
+use Amp\Uri\Uri;
 
 final class Cluster
 {
@@ -157,7 +158,7 @@ private function open(): Promise
             $compressor = $this->detectCompressor();
 
             foreach ($this->config->hosts() as $host) {
-                $connection = new Connection($host, $this->streams, $compressor);
+                $connection = new Connection(new Uri($host), $this->streams, $compressor);
 
                 try {
                     yield $connection->open(
@@ -181,7 +182,26 @@ private function open(): Promise
      */
     private function authenticate(): Promise
     {
-        return $this->connection->send(new Request\AuthResponse($this->config->user(), $this->config->password()));
+        return call(function () {
+            $request = new Request\AuthResponse(
+                $this->config->user(),
+                $this->config->password()
+            );
+
+            /** @var Frame $frame */
+            $frame = yield $this->connection->send($request);
+
+            switch (true) {
+                case $frame instanceof Response\AuthChallenge:
+                    // TODO
+
+                    break;
+                case $frame instanceof Response\AuthSuccess:
+                    break;
+                default:
+                    throw Exception\ServerException::unexpectedFrame($frame->opcode);
+            }
+        });
     }
 
     /**

src/Cluster.php

@@ -13,19 +13,21 @@
 namespace PHPinnacle\Cassis;
 
 use function Amp\asyncCall, Amp\call;
-use function Amp\Socket\connect;
+use Amp\Socket\ClientTlsContext;
+use function Amp\Socket\connect, Amp\Socket\cryptoConnect;
 use Amp\Deferred;
 use Amp\Loop;
 use Amp\Promise;
 use Amp\Socket\ClientConnectContext;
 use Amp\Socket\Socket;
+use Amp\Uri\Uri;
 
 final class Connection
 {
     const WRITE_ROUNDS = 64;
 
     /**
-     * @var string
+     * @var Uri
      */
     private $uri;
 
@@ -75,11 +77,11 @@ final class Connection
     private $lastWrite = 0;
 
     /**
-     * @param string          $uri
-     * @param Streams         $streams
-     * @param Compressor      $compressor
+     * @param Uri        $uri
+     * @param Streams    $streams
+     * @param Compressor $compressor
      */
-    public function __construct(string $uri, Streams $streams, Compressor $compressor)
+    public function __construct(Uri $uri, Streams $streams, Compressor $compressor)
     {
         $this->uri     = $uri;
         $this->streams = $streams;
@@ -137,21 +139,29 @@ public function send(Request $request): Promise
     public function open(int $timeout, int $attempts, bool $noDelay): Promise
     {
         return call(function () use ($timeout, $attempts, $noDelay) {
-            $context = new ClientConnectContext;
+            $clientContext = new ClientConnectContext;
 
             if ($attempts > 0) {
-                $context = $context->withMaxAttempts($attempts);
+                $clientContext = $clientContext->withMaxAttempts($attempts);
             }
 
             if ($timeout > 0) {
-                $context = $context->withConnectTimeout($timeout);
+                $clientContext = $clientContext->withConnectTimeout($timeout);
             }
 
             if ($noDelay) {
-                $context = $context->withTcpNoDelay();
+                $clientContext = $clientContext->withTcpNoDelay();
             }
 
-            $this->socket = yield connect($this->uri, $context);
+            $uri = \sprintf('tcp://%s:%d', $this->uri->getHost(), $this->uri->getPort());
+
+            if ($this->uri->getScheme() === 'tls') {
+                $cryptoContext = new ClientTlsContext;
+
+                $this->socket = yield cryptoConnect($uri, $clientContext, $cryptoContext);
+            } else {
+                $this->socket = yield connect($uri, $clientContext);
+            }
 
             $this->listen();
         });
@@ -210,7 +220,8 @@ private function listen(): void
                 $this->parser->append($chunk);
 
                 while ($frame = $this->parser->parse()) {
-                    if ($frame->stream === -1) {
+                    if ($frame->opcode === Frame::OPCODE_EVENT) {
+                        /** @var Response\Event $frame */
                         $this->emitter->emit($frame);
 
                         continue 2;

src/Connection.php

@@ -172,7 +172,8 @@ public function consistencyEachQuorum(): self
      */
     public function consistencySerial(): self
     {
-        $this->consistency = self::CONSISTENCY_SERIAL;
+        $this->flags |= self::FLAG_SERIAL_CONSISTENCY;
+        $this->serialConsistency = self::CONSISTENCY_SERIAL;
 
         return $this;
     }
@@ -182,7 +183,8 @@ public function consistencySerial(): self
      */
     public function consistencyLocalSerial(): self
     {
-        $this->consistency = self::CONSISTENCY_LOCAL_SERIAL;
+        $this->flags |= self::FLAG_SERIAL_CONSISTENCY;
+        $this->serialConsistency = self::CONSISTENCY_LOCAL_SERIAL;
 
         return $this;
     }
@@ -233,19 +235,6 @@ public function offset(string $offset): self
         return $this;
     }
 
-    /**
-     * @param int $consistency
-     *
-     * @return self
-     */
-    public function serialConsistency(int $consistency): self
-    {
-        $this->flags |= self::FLAG_SERIAL_CONSISTENCY;
-        $this->serialConsistency = $consistency;
-
-        return $this;
-    }
-
     /**
      * @param int $timestamp
      *

src/Context.php

@@ -14,4 +14,8 @@
 
 class ServerException extends CassisException
 {
+    public static function unexpectedFrame(int $code): self
+    {
+        return new self("Unexpected frame with opcode {$code}.");
+    }
 }

src/Exception/ServerException.php

@@ -137,7 +137,7 @@ private function frame(int $opcode, int $length): Frame
             case Frame::OPCODE_AUTHENTICATE:
                 return new Response\Authenticate($this->frameBuffer->consumeString());
             case Frame::OPCODE_AUTH_SUCCESS:
-                return new Response\AuthSuccess;
+                return new Response\AuthSuccess($this->frameBuffer->consume($length));
             case Frame::OPCODE_RESULT:
                 return new Response\Result($this->frameBuffer->consumeInt(), $this->frameBuffer->consume($length - 4));
             case Frame::OPCODE_ERROR:

src/Parser.php

@@ -45,8 +45,11 @@ public function __construct(string $user, string $password)
     public function write(Buffer $buffer): void
     {
         $buffer
-            ->appendString($this->user)
-            ->appendString($this->password)
+            ->appendUint(\strlen($this->user) + \strlen($this->password) + 2)
+            ->appendByte(0)
+            ->append($this->user)
+            ->appendByte(0)
+            ->append($this->password)
         ;
     }
 }

src/Request/AuthResponse.php

@@ -0,0 +1,20 @@
+<?php
+/**
+ * This file is part of PHPinnacle/Cassis.
+ *
+ * (c) PHPinnacle Team <dev@phpinnacle.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types = 1);
+
+namespace PHPinnacle\Cassis\Response;
+
+use PHPinnacle\Cassis\Response;
+
+final class AuthChallenge extends Response
+{
+    public $opcode = self::OPCODE_AUTH_CHALLENGE;
+}