From d9bf2a443730f601f131797f42594521426a8788 Mon Sep 17 00:00:00 2001
From: Konstantin Grachev <me@grachevko.ru>
Date: Mon, 25 Jan 2021 03:19:54 +0300
Subject: [PATCH] Add ClientConfig and ConnectionConfig, Feature Negotiation
 and Authentication

---
 README.md                                |  10 +-
 composer.json                            |   1 +
 src/Config/ClientConfig.php              | 115 +++++++++++++++++++++++
 src/Config/ConnectionConfig.php          |  81 ++++++++++++++++
 src/Connection.php                       |  70 +++++++-------
 src/Consumer.php                         |   5 -
 src/Exception/AuthenticationRequired.php |  11 +++
 src/Response.php                         |  17 ++++
 src/Subscriber.php                       |  23 +----
 tests/NsqTest.php                        |  18 +++-
 tests/SubscriberTest.php                 |  45 ---------
 11 files changed, 277 insertions(+), 119 deletions(-)
 create mode 100644 src/Config/ClientConfig.php
 create mode 100644 src/Config/ConnectionConfig.php
 create mode 100644 src/Exception/AuthenticationRequired.php
 delete mode 100644 tests/SubscriberTest.php

diff --git a/README.md b/README.md
index 79c89ba..50401cd 100644
--- a/README.md
+++ b/README.md
@@ -30,13 +30,13 @@ Features
 
 - [x] PUB
 - [x] SUB
-- [ ] Feature Negotiation	
+- [X] Feature Negotiation	
 - [ ] Discovery	
 - [ ] Backoff	
 - [ ] TLS	
 - [ ] Snappy	
 - [X] Sampling	
-- [ ] AUTH
+- [X] AUTH
 
 Usage
 -----
@@ -71,7 +71,7 @@ use Nsq\Subscriber;
 $consumer = new Consumer('tcp://nsqd:4150');
 $subscriber = new Subscriber($consumer);
 
-$generator = $subscriber->subscribe('topic', 'channel', timeout: 5);
+$generator = $subscriber->subscribe('topic', 'channel');
 foreach ($generator as $message) {
     if ($message instanceof Message) {
         $payload = $message->body;
@@ -89,10 +89,6 @@ foreach ($generator as $message) {
     // We can also communicate with generator through send
     // for example:
 
-    // Dynamically change timeout 
-    $generator->send(Subscriber::CHANGE_TIMEOUT);
-    $generator->send(10.0); // float required
-
     // Gracefully close connection (loop will be ended)
     $generator->send(Subscriber::STOP); 
 }
diff --git a/composer.json b/composer.json
index a63cd8b..e0a5d61 100644
--- a/composer.json
+++ b/composer.json
@@ -23,6 +23,7 @@
         "ergebnis/composer-normalize": "9999999-dev",
         "friendsofphp/php-cs-fixer": "^2.18",
         "infection/infection": "^0.20.2",
+        "nyholm/nsa": "^1.2",
         "phpstan/phpstan": "^0.12.68",
         "phpstan/phpstan-phpunit": "^0.12.17",
         "phpstan/phpstan-strict-rules": "^0.12.9",
diff --git a/src/Config/ClientConfig.php b/src/Config/ClientConfig.php
new file mode 100644
index 0000000..df5b9fc
--- /dev/null
+++ b/src/Config/ClientConfig.php
@@ -0,0 +1,115 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Config;
+
+use Composer\InstalledVersions;
+use InvalidArgumentException;
+use JsonSerializable;
+use function gethostname;
+
+/**
+ * This class is used for configuring the clients for nsq. Immutable properties must be set when creating the object and
+ * are sent to NSQ for feature specification or negotiation. Keep in mind that some features might require some
+ * configuration on the server-side and could be not available.
+ *
+ * @psalm-immutable
+ */
+final class ClientConfig implements JsonSerializable
+{
+    /** @psalm-suppress ImpureFunctionCall */
+    public function __construct(
+        /*
+         * The secret used for authorization, if the server requires it. This value will be ignored if the server
+         * does not require authorization.
+         */
+        public ?string $authSecret = null,
+
+        // The timeout for establishing a connection in seconds.
+        public int $connectTimeout = 10,
+
+        // An identifier used to disambiguate this client (i.e. something specific to the consumer)
+        public string $clientId = '',
+
+        // Enable deflate compression for this connection. A client cannot enable both [snappy] and [deflate].
+        public bool $deflate = false,
+        /*
+         * Configure the deflate compression level for this connection.
+         *
+         * Valid range: `1 <= deflate_level <= configured_max`
+         *
+         * Higher values mean better compression but more CPU usage for nsqd.
+         */
+        public int $deflateLevel = 6,
+
+        /*
+         * Milliseconds between heartbeats.
+         *
+         * Valid range: `1000 <= heartbeat_interval <= configured_max` (`-1` disables heartbeats)
+         */
+        public int $heartbeatInterval = 30000,
+
+        // The hostname where the client is deployed
+        public string $hostname = '',
+
+        // Configure the server-side message timeout in milliseconds for messages delivered to this client.
+        public int $msgTimeout = 60000,
+
+        /*
+         * The sample rate for incoming data to deliver a percentage of all messages received to this connection.
+         * This only applies to subscribing connections. The valid range is between 0 and 99, where 0 means that all
+         * data is sent (this is the default). 1 means that 1% of the data is sent.
+         */
+        public int $sampleRate = 0,
+
+        /*
+         * Boolean used to indicate that the client supports feature negotiation. If the server is capable,
+         * it will send back a JSON payload of supported features and metadata.
+         */
+        public bool $featureNegotiation = true,
+
+        // Enable TLS for this connection
+        public bool $tls = false,
+
+        // Enable snappy compression for this connection. A client cannot enable both [snappy] and [deflate].
+        public bool $snappy = false,
+
+        // The read timeout for connection sockets and for awaiting responses from nsq.
+        public int $readTimeout = 5,
+
+        // A string identifying the agent for this client in the spirit of HTTP.
+        public string $userAgent = '',
+    ) {
+        if ('' === $this->hostname) {
+            $this->hostname = (static fn (mixed $h): string => \is_string($h) ? $h : '')(gethostname());
+        }
+
+        if ('' === $this->userAgent) {
+            $this->userAgent = 'nsqphp/'.InstalledVersions::getPrettyVersion('nsq/nsq');
+        }
+
+        if ($this->snappy && $this->deflate) {
+            throw new InvalidArgumentException('Client cannot enable both [snappy] and [deflate]');
+        }
+    }
+
+    /**
+     * @phpstan-ignore-next-line
+     */
+    public function jsonSerialize(): array
+    {
+        return [
+            'client_id' => $this->clientId,
+            'deflate' => $this->deflate,
+            'deflate_level' => $this->deflateLevel,
+            'feature_negotiation' => $this->featureNegotiation,
+            'heartbeat_interval' => $this->heartbeatInterval,
+            'hostname' => $this->hostname,
+            'msg_timeout' => $this->msgTimeout,
+            'sample_rate' => $this->sampleRate,
+            'tls_v1' => $this->tls,
+            'user_agent' => $this->userAgent,
+        ];
+    }
+}
diff --git a/src/Config/ConnectionConfig.php b/src/Config/ConnectionConfig.php
new file mode 100644
index 0000000..7145b27
--- /dev/null
+++ b/src/Config/ConnectionConfig.php
@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Config;
+
+/**
+ * The configuration object that holds the config status for a single Connection.
+ *
+ * @psalm-immutable
+ */
+final class ConnectionConfig
+{
+    public function __construct(
+        // Whether or not authorization is required by nsqd.
+        public bool $authRequired,
+
+        // Whether deflate compression is enabled for this connection or not.
+        public bool $deflate,
+        // The deflate level. This value can be ignored if [deflate] is `false`.
+        public int $deflateLevel,
+
+        // The maximum deflate level supported by the server.
+        public int $maxDeflateLevel,
+
+        // The maximum value for message timeout.
+        public int $maxMsgTimeout,
+        /*
+         * Each nsqd is configurable with a max-rdy-count. If the consumer sends a RDY count that is outside
+         * of the acceptable range its connection will be forcefully closed.
+         */
+        public int $maxRdyCount,
+
+        // The effective message timeout.
+        public int $msgTimeout,
+
+        // The size in bytes of the buffer nsqd will use when writing to this client.
+        public int $outputBufferSize,
+        // The timeout after which any data that nsqd has buffered will be flushed to this client.
+        public int $outputBufferTimeout,
+
+        /*
+         * The sample rate for incoming data to deliver a percentage of all messages received to this connection.
+         * This only applies to subscribing connections. The valid range is between 0 and 99, where 0 means that all
+         * data is sent (this is the default). 1 means that 1% of the data is sent.
+         */
+        public int $sampleRate,
+
+        // Whether snappy compression is enabled for this connection or not.
+        public bool $snappy,
+
+        // Whether TLS is enabled for this connection or not.
+        public bool $tls,
+
+        // The nsqd version.
+        public string $version,
+    ) {
+    }
+
+    /**
+     * @phpstan-ignore-next-line
+     */
+    public static function fromArray(array $array): self
+    {
+        return new self(
+            authRequired: $array['auth_required'],
+            deflate: $array['deflate'],
+            deflateLevel: $array['deflate_level'],
+            maxDeflateLevel: $array['max_deflate_level'],
+            maxMsgTimeout: $array['max_msg_timeout'],
+            maxRdyCount: $array['max_rdy_count'],
+            msgTimeout: $array['msg_timeout'],
+            outputBufferSize: $array['output_buffer_size'],
+            outputBufferTimeout: $array['output_buffer_timeout'],
+            sampleRate: $array['sample_rate'],
+            snappy: $array['snappy'],
+            tls: $array['tls_v1'],
+            version: $array['version'],
+        );
+    }
+}
diff --git a/src/Connection.php b/src/Connection.php
index 69edcb6..95ed223 100644
--- a/src/Connection.php
+++ b/src/Connection.php
@@ -4,7 +4,9 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Composer\InstalledVersions;
+use Nsq\Config\ClientConfig;
+use Nsq\Config\ConnectionConfig;
+use Nsq\Exception\AuthenticationRequired;
 use Nsq\Exception\ConnectionFail;
 use Nsq\Exception\UnexpectedResponse;
 use Nsq\Reconnect\ExponentialStrategy;
@@ -17,6 +19,7 @@ use Socket\Raw\Exception;
 use Socket\Raw\Factory;
 use Socket\Raw\Socket;
 use function addcslashes;
+use function http_build_query;
 use function implode;
 use function json_encode;
 use function pack;
@@ -37,38 +40,21 @@ abstract class Connection
 
     private ReconnectStrategy $reconnect;
 
-    /**
-     * @var array{
-     *             client_id: string,
-     *             hostname: string,
-     *             user_agent: string,
-     *             heartbeat_interval: int|null,
-     *             }
-     */
-    private array $features;
+    private ClientConfig $clientConfig;
+
+    private ?ConnectionConfig $connectionConfig = null;
 
     public function __construct(
         string $address,
-        string $clientId = null,
-        string $hostname = null,
-        string $userAgent = null,
-        int $heartbeatInterval = null,
-        int $sampleRate = 0,
+        ClientConfig $clientConfig = null,
         ReconnectStrategy $reconnectStrategy = null,
         LoggerInterface $logger = null,
     ) {
         $this->address = $address;
 
-        $this->features = [
-            'client_id' => $clientId ?? '',
-            'hostname' => $hostname ?? (static fn (mixed $h): string => \is_string($h) ? $h : '')(gethostname()),
-            'user_agent' => $userAgent ?? 'nsqphp/'.InstalledVersions::getPrettyVersion('nsq/nsq'),
-            'heartbeat_interval' => $heartbeatInterval,
-            'sample_rate' => $sampleRate,
-        ];
-
         $this->logger = $logger ?? new NullLogger();
         $this->reconnect = $reconnectStrategy ?? new ExponentialStrategy(logger: $this->logger);
+        $this->clientConfig = $clientConfig ?? new ClientConfig();
     }
 
     public function connect(): void
@@ -87,9 +73,27 @@ abstract class Connection
 
             $this->socket->write('  V2');
 
-            $body = json_encode($this->features, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
+            $body = json_encode($this->clientConfig, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
 
-            $this->command('IDENTIFY', data: $body)->response()->okOrFail();
+            $response = $this->command('IDENTIFY', data: $body)->response();
+
+            if ($this->clientConfig->featureNegotiation) {
+                $this->connectionConfig = ConnectionConfig::fromArray($response->toArray());
+            }
+
+            if ($this->connectionConfig->snappy || $this->connectionConfig->deflate) {
+                $this->response()->okOrFail();
+            }
+
+            if ($this->connectionConfig->authRequired) {
+                if (null === $this->clientConfig->authSecret) {
+                    throw new AuthenticationRequired('NSQ requires authorization, set ClientConfig::$authSecret before connecting');
+                }
+
+                $authResponse = $this->command('AUTH', data: $this->clientConfig->authSecret)->response()->toArray();
+
+                $this->logger->info('Authorization response: '.http_build_query($authResponse));
+            }
         });
     }
 
@@ -120,16 +124,6 @@ abstract class Connection
         return null !== $this->socket;
     }
 
-    /**
-     * @psalm-suppress PossiblyFalseOperand
-     */
-    protected function auth(string $secret): string
-    {
-        $size = pack('N', \strlen($secret));
-
-        return 'AUTH'.PHP_EOL.$size.$secret;
-    }
-
     /**
      * @param array<int, int|string>|string $params
      */
@@ -176,9 +170,11 @@ abstract class Connection
         // @codeCoverageIgnoreEnd
     }
 
-    public function receive(float $timeout = 0): ?Response
+    public function receive(float $timeout = null): ?Response
     {
         $socket = $this->socket();
+
+        $timeout ??= $this->clientConfig->readTimeout;
         $deadline = microtime(true) + $timeout;
 
         if (!$this->hasMessage($timeout)) {
@@ -226,7 +222,7 @@ abstract class Connection
 
     protected function response(): Response
     {
-        return $this->receive(1) ?? throw UnexpectedResponse::null();
+        return $this->receive() ?? throw UnexpectedResponse::null();
     }
 
     private function socket(): Socket
diff --git a/src/Consumer.php b/src/Consumer.php
index 6859ede..6a67901 100644
--- a/src/Consumer.php
+++ b/src/Consumer.php
@@ -30,11 +30,6 @@ final class Consumer extends Connection
         $this->rdy = $count;
     }
 
-    public function consume(float $timeout): ?Message
-    {
-        return $this->receive($timeout)?->toMessage($this);
-    }
-
     /**
      * Finish a message (indicate successful processing).
      */
diff --git a/src/Exception/AuthenticationRequired.php b/src/Exception/AuthenticationRequired.php
new file mode 100644
index 0000000..0c493a5
--- /dev/null
+++ b/src/Exception/AuthenticationRequired.php
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Exception;
+
+use RuntimeException;
+
+final class AuthenticationRequired extends RuntimeException implements NsqException
+{
+}
diff --git a/src/Response.php b/src/Response.php
index aa29f38..a9e2d4d 100644
--- a/src/Response.php
+++ b/src/Response.php
@@ -7,6 +7,9 @@ namespace Nsq;
 use Nsq\Exception\NsqError;
 use Nsq\Exception\UnexpectedResponse;
 use PHPinnacle\Buffer\ByteBuffer;
+use function json_decode;
+use function sprintf;
+use const JSON_THROW_ON_ERROR;
 
 final class Response
 {
@@ -50,6 +53,20 @@ final class Response
         return self::TYPE_RESPONSE === $this->type && self::HEARTBEAT === $this->buffer->bytes();
     }
 
+    /**
+     * @phpstan-ignore-next-line
+     */
+    public function toArray(): array
+    {
+        if (self::TYPE_RESPONSE !== $this->type) {
+            // @codeCoverageIgnoreStart
+            throw new UnexpectedResponse(sprintf('"%s" type expected, but "%s" received.', self::TYPE_RESPONSE, $this->type));
+            // @codeCoverageIgnoreEnd
+        }
+
+        return json_decode($this->buffer->bytes(), true, flags: JSON_THROW_ON_ERROR);
+    }
+
     public function toMessage(Consumer $reader): Message
     {
         if (self::TYPE_MESSAGE !== $this->type) {
diff --git a/src/Subscriber.php b/src/Subscriber.php
index cda1a69..5707484 100644
--- a/src/Subscriber.php
+++ b/src/Subscriber.php
@@ -5,15 +5,10 @@ declare(strict_types=1);
 namespace Nsq;
 
 use Generator;
-use InvalidArgumentException;
-use function get_debug_type;
-use function sprintf;
 
 final class Subscriber
 {
     public const STOP = 0;
-    public const CHANGE_TIMEOUT = 1;
-    public const TIMEOUT = 2;
 
     private Consumer $reader;
 
@@ -25,32 +20,18 @@ final class Subscriber
     /**
      * @psalm-return Generator<int, Message|float|null, int|float|null, void>
      */
-    public function subscribe(string $topic, string $channel, float $timeout = 0): Generator
+    public function subscribe(string $topic, string $channel): Generator
     {
         $this->reader->sub($topic, $channel);
 
         while (true) {
             $this->reader->rdy(1);
 
-            $command = yield $this->reader->consume($timeout);
+            $command = yield $this->reader->receive()?->toMessage($this->reader);
 
             if (self::STOP === $command) {
                 break;
             }
-
-            if (self::CHANGE_TIMEOUT === $command) {
-                $newTimeout = yield null;
-
-                if (!\is_float($newTimeout)) {
-                    throw new InvalidArgumentException(sprintf('Timeout must be float, "%s" given.', get_debug_type($newTimeout)));
-                }
-
-                $timeout = $newTimeout;
-            }
-
-            if (self::TIMEOUT === $command) {
-                yield $timeout;
-            }
         }
 
         $this->reader->disconnect();
diff --git a/tests/NsqTest.php b/tests/NsqTest.php
index 6c3a086..18a38d8 100644
--- a/tests/NsqTest.php
+++ b/tests/NsqTest.php
@@ -2,10 +2,12 @@
 
 declare(strict_types=1);
 
+use Nsq\Config\ClientConfig;
 use Nsq\Consumer;
 use Nsq\Message;
 use Nsq\Producer;
 use Nsq\Subscriber;
+use Nyholm\NSA;
 use PHPUnit\Framework\TestCase;
 
 final class NsqTest extends TestCase
@@ -17,10 +19,13 @@ final class NsqTest extends TestCase
 
         $consumer = new Consumer(
             address: 'tcp://localhost:4150',
-            heartbeatInterval: 1000,
+            clientConfig: new ClientConfig(
+                heartbeatInterval: 1000,
+                readTimeout: 1,
+            ),
         );
         $subscriber = new Subscriber($consumer);
-        $generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__, 1);
+        $generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__);
 
         /** @var null|Message $message */
         $message = $generator->current();
@@ -65,8 +70,13 @@ final class NsqTest extends TestCase
         $message = $generator->current();
         self::assertNull($message);
 
-        $generator->send(Subscriber::CHANGE_TIMEOUT);
-        $generator->send(10.0);
+        NSA::setProperty(
+            NSA::getProperty($consumer, 'clientConfig'),
+            'readTimeout',
+            10,
+        );
+
+        $generator->next();
 
         /** @var null|Message $message */
         $message = $generator->current();
diff --git a/tests/SubscriberTest.php b/tests/SubscriberTest.php
deleted file mode 100644
index c413ed1..0000000
--- a/tests/SubscriberTest.php
+++ /dev/null
@@ -1,45 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-use Nsq\Consumer;
-use Nsq\Subscriber;
-use PHPUnit\Framework\TestCase;
-
-final class SubscriberTest extends TestCase
-{
-    private Subscriber $subscriber;
-
-    protected function setUp(): void
-    {
-        $consumer = new Consumer(
-            address: 'tcp://localhost:4150',
-        );
-
-        $this->subscriber = new Subscriber($consumer);
-    }
-
-    public function testChangeInterval(): void
-    {
-        $generator = $this->subscriber->subscribe(__FUNCTION__, __FUNCTION__, 0.1);
-
-        self::assertSame(0.1, $generator->send(Subscriber::TIMEOUT));
-        $generator->next();
-
-        $generator->send(Subscriber::CHANGE_TIMEOUT);
-        $generator->send(0.2);
-
-        self::assertSame(0.2, $generator->send(Subscriber::TIMEOUT));
-    }
-
-    public function testInvalidChangeInterval(): void
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('Timeout must be float, "string" given.');
-
-        $generator = $this->subscriber->subscribe(__FUNCTION__, __FUNCTION__);
-        $generator->send(Subscriber::CHANGE_TIMEOUT);
-        // @phpstan-ignore-next-line
-        $generator->send('bla');
-    }
-}