Snappy

.gitattributes

@@ -1,21 +0,0 @@
-# Exclude build/test files from archive
-/.editorconfig          export-ignore
-/.gitattributes         export-ignore
-/.github                export-ignore
-/.gitignore             export-ignore
-/.php_cs                export-ignore
-/.php_cs.dist           export-ignore
-/.psalm                 export-ignore
-/docs                   export-ignore
-/examples               export-ignore
-/infection.json         export-ignore
-/infection.json.dist    export-ignore
-/phpstan.neon           export-ignore
-/phpunit.xml            export-ignore
-/phpunit.xml.dist       export-ignore
-/psalm.xml              export-ignore
-/tests                  export-ignore
-
-# Configure diff output for .php and .phar files.
-*.php diff=php
-*.phar -diff

.github/workflows/ci.yaml

@@ -33,7 +33,6 @@ jobs:
                 with:
                     php-version: ${{ matrix.php }}
                     coverage: pcov
-                    extensions: kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -59,9 +58,6 @@ jobs:
                 run: composer update --no-progress --no-interaction --prefer-dist --prefer-lowest
                 if: ${{ matrix.dependencies == 'lowest' }}
 
-            -   name: Install nsq bin
-                run: curl -L https://github.com/nsqio/nsq/releases/download/v1.2.0/nsq-1.2.0.linux-amd64.go1.12.9.tar.gz | tar xz --strip 1
-
             -   name: Run tests
                 run: vendor/bin/phpunit --coverage-clover=build/coverage-report.xml
 
@@ -112,7 +108,6 @@ jobs:
                 uses: shivammathur/setup-php@v2
                 with:
                     php-version: '8.0'
-                    extensions: snappy-kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -144,7 +139,6 @@ jobs:
                 uses: shivammathur/setup-php@v2
                 with:
                     php-version: '8.0'
-                    extensions: snappy-kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -200,9 +194,6 @@ jobs:
             -   name: Install dependencies
                 run: composer update --no-progress --no-interaction --prefer-dist
 
-            -   name: Install nsq bin
-                run: curl -L https://github.com/nsqio/nsq/releases/download/v1.2.0/nsq-1.2.0.linux-amd64.go1.12.9.tar.gz | tar xz --strip 1
-
             -   name: Run script
                 env:
                     STRYKER_DASHBOARD_API_KEY: ${{ secrets.STRYKER_DASHBOARD_API_KEY }}

.gitignore

@@ -4,14 +4,3 @@
 /.php_cs.cache
 /.phpunit.result.cache
 /infection.log
-
-# Nsq
-bin/nsq_stat
-bin/nsq_tail
-bin/nsq_to_file
-bin/nsq_to_http
-bin/nsq_to_nsq
-bin/nsqadmin
-bin/nsqd
-bin/nsqlookupd
-bin/to_nsq

.php_cs.dist

@@ -8,19 +8,14 @@ return (new PhpCsFixer\Config())
         '@PhpCsFixer:risky' => true,
         '@PSR12' => true,
         '@PSR12:risky' => true,
-        'braces' => [
-            'allow_single_line_closure' => true,
-        ],
-        'blank_line_before_statement' => [
-            'statements' => ['continue', 'do', 'die', 'exit', 'goto', 'if', 'return', 'switch', 'throw', 'try'],
-        ],
         'declare_strict_types' => true,
-        'global_namespace_import' => ['import_classes' => false, 'import_constants' => false, 'import_functions' => false],
         'php_unit_internal_class' => false,
-        'php_unit_test_case_static_method_calls'=> ['call_type' => 'self'],
         'php_unit_test_class_requires_covers' => false,
-        'phpdoc_to_comment' => false,
         'yoda_style' => true,
+        'php_unit_test_case_static_method_calls'=> ['call_type' => 'self'],
+        'blank_line_before_statement' => [
+            'statements' => ['continue', 'do', 'die', 'exit', 'goto', 'if', 'return', 'switch', 'throw', 'try']
+        ],
     ])
     ->setFinder(
         PhpCsFixer\Finder::create()

README.md

@@ -1,6 +1,6 @@
 # Nsq PHP
 
-<img src="https://github.com/nsqphp/nsqphp/raw/main/docs/logo.png" alt="" align="left" width="150">
+<img src="https://github.com/nsqphp/nsqphp/raw/main/logo.png" alt="" align="left" width="150">
 
 PHP Client for [NSQ](https://nsq.io/).
 
@@ -34,50 +34,64 @@ Features
 - [ ] Discovery	
 - [ ] Backoff	
 - [ ] TLS	
-- [ ] Deflate	
+- [ ] Snappy	
-- [X] Snappy	
 - [X] Sampling	
 - [X] AUTH
 
 Usage
 -----
 
-### Producer
+### Publish
 
 ```php
 use Nsq\Producer;
 
-$producer = Producer::create(address: 'tcp://nsqd:4150');
+$producer = new Producer(address: 'tcp://nsqd:4150');
 
 // Publish a message to a topic
-$producer->publish('topic', 'Simple message');
+$producer->pub('topic', 'Simple message');
 
 // Publish multiple messages to a topic (atomically) 
-$producer->publish('topic', [
+$producer->mpub('topic', [
     'Message one',
     'Message two',
 ]);
 
 // Publish a deferred message to a topic
-$producer->publish('topic', 'Deferred message', delay: 5000);
+$producer->dpub('topic', 'Deferred message', delay: 5000);
 ```
 
-### Consumer
+### Subscription
 
 ```php
 use Nsq\Consumer;
 use Nsq\Message;
+use Nsq\Subscriber;
 
-$consumer = Consumer::create(
+$consumer = new Consumer('tcp://nsqd:4150');
-    address: 'tcp://nsqd:4150', 
+$subscriber = new Subscriber($consumer);
-    topic: 'topic',
+
-    channel: 'channel',
+$generator = $subscriber->subscribe('topic', 'channel');
-    onMessage: static function (Message $message): Generator {
+foreach ($generator as $message) {
-        yield $message->touch(); // Reset the timeout for an in-flight message        
+    if ($message instanceof Message) {
-        yield $message->requeue(timeout: 5000); // Re-queue a message (indicate failure to process)        
+        $payload = $message->body;
-        yield $message->finish(); // Finish a message (indicate successful processing)        
+
-    },
+        // handle message
-);
+
+        $message->touch(); // Reset the timeout for an in-flight message        
+        $message->requeue(timeout: 5000); // Re-queue a message (indicate failure to process)        
+        $message->finish(); // Finish a message (indicate successful processing)        
+    }
+    
+    // In case of nothing received during timeout generator will return NULL
+    // Here we can do something between messages, like pcntl_signal_dispatch()
+    
+    // We can also communicate with generator through send
+    // for example:
+
+    // Gracefully close connection (loop will be ended)
+    $generator->send(Subscriber::STOP); 
+}
 ```
 
 ### Integrations

composer.json

@@ -13,13 +13,12 @@
     "require": {
         "php": "^8.0.1",
         "ext-json": "*",
-        "amphp/socket": "^1.1",
+        "clue/socket-raw": "^1.5",
         "composer/semver": "^3.2",
         "phpinnacle/buffer": "^1.2",
         "psr/log": "^1.1"
     },
     "require-dev": {
-        "amphp/log": "^1.1",
         "dg/bypass-finals": "^1.3",
         "ergebnis/composer-normalize": "9999999-dev",
         "friendsofphp/php-cs-fixer": "^2.18",
@@ -43,7 +42,7 @@
     "prefer-stable": true,
     "scripts": {
         "cs": [
-            "vendor/bin/php-cs-fixer fix --using-cache=no"
+            "vendor/bin/php-cs-fixer fix"
         ],
         "cs-check": [
             "vendor/bin/php-cs-fixer fix --verbose --diff --dry-run"
@@ -60,7 +59,7 @@
             "vendor/bin/psalm"
         ],
         "test": [
-            "@norm",
+            "@norm-check",
             "@cs",
             "@phpstan",
             "@psalm",

docker-compose.yml

@@ -3,22 +3,13 @@ version: '3.7'
 services:
     nsqd:
         image: nsqio/nsq:v1.2.0
-        labels:
+        command: /nsqd
-            ru.grachevko.dhu: 'nsqd'
-        command: /nsqd -log-level debug
-#        command: /nsqd
         ports:
             - 4150:4150
             - 4151:4151
 
     nsqadmin:
         image: nsqio/nsq:v1.2.0
-        labels:
-            ru.grachevko.dhu: 'nsqadmin'
         command: /nsqadmin --nsqd-http-address=nsqd:4151 --http-address=0.0.0.0:4171
         ports:
             - 4171:4171
-
-    tail:
-        image: nsqio/nsq:v1.2.0
-        command: nsq_tail -channel nsq_tail -topic local -nsqd-tcp-address nsqd:4150

examples/consumer.php

@@ -1,43 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-require __DIR__.'/../vendor/autoload.php';
-
-use Amp\ByteStream;
-use Amp\Log\ConsoleFormatter;
-use Amp\Log\StreamHandler;
-use Amp\Loop;
-use Amp\Promise;
-use Monolog\Logger;
-use Monolog\Processor\PsrLogMessageProcessor;
-use Nsq\Config\ClientConfig;
-use Nsq\Consumer;
-use Nsq\Message;
-use function Amp\call;
-
-Loop::run(static function () {
-    $handler = new StreamHandler(ByteStream\getStdout());
-    $handler->setFormatter(new ConsoleFormatter());
-    $logger = new Logger('publisher', [$handler], [new PsrLogMessageProcessor()]);
-
-    $consumer = new Consumer(
-        'tcp://localhost:4150',
-        topic: 'local',
-        channel: 'local',
-        onMessage: static function (Message $message) use ($logger): Promise {
-            return call(function () use ($message, $logger): Generator {
-                $logger->info('Received: {body}', ['body' => $message->body]);
-
-                yield $message->finish();
-            });
-        },
-        clientConfig: new ClientConfig(
-            deflate: false,
-            snappy: true,
-        ),
-        logger: $logger,
-    );
-
-    yield $consumer->connect();
-});

examples/producer.php

@@ -1,36 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-require __DIR__.'/../vendor/autoload.php';
-
-use Amp\ByteStream;
-use Amp\Log\ConsoleFormatter;
-use Amp\Log\StreamHandler;
-use Amp\Loop;
-use Monolog\Logger;
-use Monolog\Processor\PsrLogMessageProcessor;
-use Nsq\Config\ClientConfig;
-use Nsq\Producer;
-
-Loop::run(static function () {
-    $handler = new StreamHandler(ByteStream\getStdout());
-    $handler->setFormatter(new ConsoleFormatter());
-    $logger = new Logger('publisher', [$handler], [new PsrLogMessageProcessor()]);
-
-    $producer = new Producer(
-        'tcp://localhost:4150',
-        clientConfig: new ClientConfig(
-            deflate: false,
-            heartbeatInterval: 5000,
-            snappy: true,
-        ),
-        logger: $logger,
-    );
-
-    yield $producer->connect();
-
-    while (true) {
-        yield $producer->publish(topic: 'local', body: array_fill(0, 200, 'Message body!'));
-    }
-});

src/Buffer.php

@@ -1,34 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-use PHPinnacle\Buffer\ByteBuffer;
-
-/**
- * @psalm-suppress
- */
-final class Buffer extends ByteBuffer
-{
-    public function readUInt32LE(): int
-    {
-        /** @phpstan-ignore-next-line  */
-        return unpack('V', $this->consume(4))[1];
-    }
-
-    public function consumeTimestamp(): int
-    {
-        return $this->consumeUint64();
-    }
-
-    public function consumeAttempts(): int
-    {
-        return $this->consumeUint16();
-    }
-
-    public function consumeMessageID(): string
-    {
-        return $this->consume(16);
-    }
-}

src/Command.php

@@ -1,111 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-use PHPinnacle\Buffer\ByteBuffer;
-
-/**
- * @internal
- */
-final class Command
-{
-    public static function magic(): string
-    {
-        return '  V2';
-    }
-
-    public static function identify(string $data): string
-    {
-        return self::pack('IDENTIFY', data: $data);
-    }
-
-    public static function auth(?string $authSecret): string
-    {
-        return self::pack('AUTH', data: $authSecret);
-    }
-
-    public static function nop(): string
-    {
-        return self::pack('NOP');
-    }
-
-    public static function cls(): string
-    {
-        return self::pack('CLS');
-    }
-
-    public static function rdy(int $count): string
-    {
-        return self::pack('RDY', (string) $count);
-    }
-
-    public static function fin(string $id): string
-    {
-        return self::pack('FIN', $id);
-    }
-
-    public static function req(string $id, int $timeout): string
-    {
-        return self::pack('REQ', [$id, $timeout]);
-    }
-
-    public static function touch(string $id): string
-    {
-        return self::pack('TOUCH', $id);
-    }
-
-    public static function pub(string $topic, string $body): string
-    {
-        return self::pack('PUB', $topic, $body);
-    }
-
-    /**
-     * @param array<int, string> $bodies
-     */
-    public static function mpub(string $topic, array $bodies): string
-    {
-        static $buffer;
-        $buffer ??= new ByteBuffer();
-
-        $buffer->appendUint32(\count($bodies));
-
-        foreach ($bodies as $body) {
-            $buffer->appendUint32(\strlen($body));
-            $buffer->append($body);
-        }
-
-        return self::pack('MPUB', $topic, $buffer->flush());
-    }
-
-    public static function dpub(string $topic, string $body, int $delay): string
-    {
-        return self::pack('DPUB', [$topic, $delay], $body);
-    }
-
-    public static function sub(string $topic, string $channel): string
-    {
-        return self::pack('SUB', [$topic, $channel]);
-    }
-
-    /**
-     * @param array<int, scalar>|string $params
-     */
-    private static function pack(string $command, array | string $params = [], string $data = null): string
-    {
-        static $buffer;
-        $buffer ??= new Buffer();
-
-        $command = implode(' ', [$command, ...((array) $params)]);
-
-        $buffer->append($command.PHP_EOL);
-
-        if (null !== $data) {
-            $buffer->appendUint32(\strlen($data));
-            $buffer->append($data);
-        }
-
-        return $buffer->flush();
-    }
-}

src/Config/ClientConfig.php

@@ -5,6 +5,9 @@ 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
@@ -13,34 +16,25 @@ use Composer\InstalledVersions;
  *
  * @psalm-immutable
  */
-final class ClientConfig implements \JsonSerializable
+final class ClientConfig implements JsonSerializable
 {
-    /**
+    /** @psalm-suppress ImpureFunctionCall */
-     * @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.
-         * 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)
-         * 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].
-         * 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`
@@ -49,54 +43,42 @@ final class ClientConfig implements \JsonSerializable
          */
         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
-         * The hostname where the client is deployed.
-         */
         public string $hostname = '',
 
-        /**
+        // Configure the server-side message timeout in milliseconds for messages delivered to this client.
-         * 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
-         * Enable TLS for this connection.
-         */
         public bool $tls = false,
 
-        /**
+        // Enable snappy compression for this connection. A client cannot enable both [snappy] and [deflate].
-         * 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.
-         * 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.
-         * A string identifying the agent for this client in the spirit of HTTP.
-         */
         public string $userAgent = '',
     ) {
         $this->featureNegotiation = true; // Always enabled
@@ -110,7 +92,7 @@ final class ClientConfig implements \JsonSerializable
         }
 
         if ($this->snappy && $this->deflate) {
-            throw new \InvalidArgumentException('Client cannot enable both [snappy] and [deflate]');
+            throw new InvalidArgumentException('Client cannot enable both [snappy] and [deflate]');
         }
     }
 
@@ -133,9 +115,4 @@ final class ClientConfig implements \JsonSerializable
             'user_agent' => $this->userAgent,
         ];
     }
-
-    public function toString(): string
-    {
-        return json_encode($this, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
-    }
 }

src/Config/ConnectionConfig.php

@@ -12,72 +12,47 @@ namespace Nsq\Config;
 final class ConnectionConfig
 {
     public function __construct(
-        /**
+        // Whether or not authorization is required by nsqd.
-         * Whether or not authorization is required by nsqd.
-         */
         public bool $authRequired,
 
-        /**
+        // Whether deflate compression is enabled for this connection or not.
-         * 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`.
-        /**
-         * The deflate level. This value can be ignored if [deflate] is `false`.
-         */
         public int $deflateLevel,
 
-        /**
+        // The maximum deflate level supported by the server.
-         * The maximum deflate level supported by the server.
-         */
         public int $maxDeflateLevel,
 
-        /**
+        // The maximum value for message timeout.
-         * 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.
-         * The effective message timeout.
-         */
         public int $msgTimeout,
 
-        /**
+        // The size in bytes of the buffer nsqd will use when writing to this client.
-         * 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.
-        /**
-         * 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.
-         * Whether snappy compression is enabled for this connection or not.
-         */
         public bool $snappy,
 
-        /**
+        // Whether TLS is enabled for this connection or not.
-         * Whether TLS is enabled for this connection or not.
-         */
         public bool $tls,
 
-        /**
+        // The nsqd version.
-         * The nsqd version.
-         */
         public string $version,
     ) {
     }

src/Connection.php

@@ -4,136 +4,340 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Amp\Promise;
 use Nsq\Config\ClientConfig;
 use Nsq\Config\ConnectionConfig;
 use Nsq\Exception\AuthenticationRequired;
-use Nsq\Exception\NsqException;
+use Nsq\Exception\ConnectionFail;
-use Nsq\Frame\Response;
+use Nsq\Exception\UnexpectedResponse;
-use Nsq\Stream\GzipStream;
+use Nsq\Reconnect\ExponentialStrategy;
-use Nsq\Stream\NullStream;
+use Nsq\Reconnect\ReconnectStrategy;
-use Nsq\Stream\SnappyStream;
+use PHPinnacle\Buffer\ByteBuffer;
-use Nsq\Stream\SocketStream;
+use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
-use function Amp\call;
+use Psr\Log\NullLogger;
+use Socket\Raw\Exception;
+use Socket\Raw\Factory;
+use Socket\Raw\Socket;
+use function addcslashes;
+use function hash;
+use function http_build_query;
+use function implode;
+use function json_encode;
+use function pack;
+use function snappy_compress;
+use function unpack;
+use const JSON_FORCE_OBJECT;
+use const JSON_THROW_ON_ERROR;
+use const PHP_EOL;
 
 /**
  * @internal
+ *
+ * @property ConnectionConfig $connectionConfig
  */
 abstract class Connection
 {
-    protected Stream $stream;
+    use LoggerAwareTrait;
+
+    private string $address;
+
+    private ?Socket $socket = null;
+
+    private ReconnectStrategy $reconnect;
+
+    private ClientConfig $clientConfig;
+
+    private ?ConnectionConfig $connectionConfig = null;
 
     public function __construct(
-        private string $address,
+        string $address,
-        private ClientConfig $clientConfig,
+        ClientConfig $clientConfig = null,
-        private LoggerInterface $logger,
+        ReconnectStrategy $reconnectStrategy = null,
+        LoggerInterface $logger = null,
     ) {
-        $this->stream = new NullStream();
+        $this->address = $address;
+
+        $this->logger = $logger ?? new NullLogger();
+        $this->reconnect = $reconnectStrategy ?? new ExponentialStrategy(logger: $this->logger);
+        $this->clientConfig = $clientConfig ?? new ClientConfig();
     }
 
-    public function __destruct()
+    public function connect(): void
     {
-        $this->close();
+        $this->reconnect->connect(function (): void {
-    }
+            try {
+                $this->socket = (new Factory())->createClient($this->address);
+            }
+            // @codeCoverageIgnoreStart
+            catch (Exception $e) {
+                $this->logger->error('Connecting to {address} failed.', ['address' => $this->address]);
 
-    /**
+                throw ConnectionFail::fromThrowable($e);
-     * @return Promise<void>
+            }
-     */
+            // @codeCoverageIgnoreEnd
-    public function connect(): Promise
-    {
-        return call(function (): \Generator {
-            $buffer = new Buffer();
 
-            /** @var SocketStream $stream */
+            $this->socket->write('  V2');
-            $stream = yield SocketStream::connect($this->address);
 
-            yield $stream->write(Command::magic());
+            $body = json_encode($this->clientConfig, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
-            yield $stream->write(Command::identify($this->clientConfig->toString()));
 
-            /** @var Response $response */
+            $response = $this->command('IDENTIFY', data: $body)->response();
-            $response = yield $this->response($stream, $buffer);
-            $connectionConfig = ConnectionConfig::fromArray($response->toArray());
 
-            if ($connectionConfig->snappy) {
+            $this->connectionConfig = ConnectionConfig::fromArray($response->toArray());
-                $stream = new SnappyStream($stream, $buffer->flush());
 
-                /** @var Response $response */
+            if ($this->connectionConfig->snappy || $this->connectionConfig->deflate) {
-                $response = yield $this->response($stream, $buffer);
+                $this->response()->okOrFail();
-
-                if (!$response->isOk()) {
-                    throw new NsqException();
-                }
             }
 
-            if ($connectionConfig->deflate) {
+            if ($this->connectionConfig->authRequired) {
-                $stream = new GzipStream($stream);
-
-                /** @var Response $response */
-                $response = yield $this->response($stream, $buffer);
-
-                if (!$response->isOk()) {
-                    throw new NsqException();
-                }
-            }
-
-            if ($connectionConfig->authRequired) {
                 if (null === $this->clientConfig->authSecret) {
-                    throw new AuthenticationRequired();
+                    throw new AuthenticationRequired('NSQ requires authorization, set ClientConfig::$authSecret before connecting');
                 }
 
-                yield $stream->write(Command::auth($this->clientConfig->authSecret));
+                $authResponse = $this->command('AUTH', data: $this->clientConfig->authSecret)->response()->toArray();
 
-                /** @var Response $response */
+                $this->logger->info('Authorization response: '.http_build_query($authResponse));
-                $response = yield $this->response($stream, $buffer);
-
-                $this->logger->info('Authorization response: '.http_build_query($response->toArray()));
             }
-
-            $this->stream = $stream;
         });
     }
 
-    public function close(): void
+    /**
+     * Cleanly close your connection (no more messages are sent).
+     */
+    public function disconnect(): void
     {
-//        $this->stream->write(Command::cls());
+        if (null === $this->socket) {
+            return;
+        }
 
-        $this->stream->close();
+        try {
-        $this->stream = new NullStream();
+            $this->socket->write('CLS'.PHP_EOL);
+            $this->socket->close();
+        }
+        // @codeCoverageIgnoreStart
+        catch (Exception $e) {
+            $this->logger->debug($e->getMessage(), ['exception' => $e]);
+        }
+        // @codeCoverageIgnoreEnd
+
+        $this->socket = null;
+        $this->connectionConfig = null;
     }
 
-    protected function handleError(Frame\Error $error): void
+    public function isReady(): bool
     {
-        $this->logger->error($error->data);
+        return null !== $this->socket;
-
-        if (ErrorType::terminable($error)) {
-            $this->close();
-
-            throw $error->toException();
-        }
     }
 
     /**
-     * @return Promise<Frame\Response>
+     * @param array<int, int|string>|string $params
      */
-    private function response(Stream $stream, Buffer $buffer): Promise
+    protected function command(string $command, array | string $params = [], string $data = null): self
     {
-        return call(function () use ($stream, $buffer): \Generator {
+        $socket = $this->socket();
-            while (true) {
+
-                $response = Parser::parse($buffer);
+        $buffer = [] === $params ? $command : implode(' ', [$command, ...((array) $params)]);
+        $buffer .= PHP_EOL;
+
+        if (null !== $data) {
+            $buffer .= pack('N', \strlen($data));
+            $buffer .= $data;
+        }
+
+        $this->logger->debug('Prepare send uncompressed buffer: {bytes}', ['bytes' => addcslashes($buffer, PHP_EOL)]);
+
+        if ($this->connectionConfig?->snappy) {
+            $identifierFrame = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
+            $compressedFrame = 0x00;
+            $uncompressedFrame = 0x01;
+
+            $chunk = snappy_compress($buffer);
+            [$chunk, $compressFrame] = match (\strlen($chunk) < \strlen($buffer)) {
+                true => [$chunk, $compressedFrame],
+                false => [$buffer, $uncompressedFrame],
+            };
+
+            $size = \strlen($chunk) + 4;
+
+            $buffer = new ByteBuffer();
+            foreach ([...$identifierFrame, $compressFrame, $size, $size >> 8, $size >> 16] as $byte) {
+                $buffer->appendUint8($byte);
+            }
+
+            $crc32c = hash('crc32c', $data, true);
+            $crc32c = unpack('V', $crc32c)[1];
+
+            $unsignedRightShift = static function ($a, $b) {
+                if ($b >= 32 || $b < -32) {
+                    $m = (int) ($b / 32);
+                    $b -= ($m * 32);
+                }
+
+                if ($b < 0) {
+                    $b = 32 + $b;
+                }
+
+                if (0 === $b) {
+                    return (($a >> 1) & 0x7fffffff) * 2 + (($a >> $b) & 1);
+                }
+
+                if ($a < 0) {
+                    $a >>= 1;
+                    $a &= 2147483647;
+                    $a |= 0x40000000;
+                    $a >>= ($b - 1);
+                } else {
+                    $a >>= $b;
+                }
+
+                return $a;
+            };
+            $checksum = $unsignedRightShift((($crc32c >> 15) | ($crc32c << 17)) + 0xa282ead8, 0);
+
+            $buffer->appendUint32($checksum);
+            $buffer->append($chunk);
+
+            $buffer = $buffer->bytes();
+        }
+
+        $this->logger->debug('Prepare send compressed buffer: {bytes}', ['bytes' => addcslashes($buffer, PHP_EOL)]);
+
+        try {
+            $socket->write($buffer);
+        }
+        // @codeCoverageIgnoreStart
+        catch (Exception $e) {
+            $this->disconnect();
+
+            $this->logger->error($e->getMessage(), ['exception' => $e]);
+
+            throw ConnectionFail::fromThrowable($e);
+        }
+        // @codeCoverageIgnoreEnd
+        return $this;
+    }
+
+    public function hasMessage(float $timeout = 0): bool
+    {
+        try {
+            return false !== $this->socket()->selectRead($timeout);
+        }
+        // @codeCoverageIgnoreStart
+        catch (Exception $e) {
+            $this->disconnect();
+
+            throw ConnectionFail::fromThrowable($e);
+        }
+        // @codeCoverageIgnoreEnd
+    }
+
+    public function receive(float $timeout = null): ?Response
+    {
+        $socket = $this->socket();
+
+        $timeout ??= $this->clientConfig->readTimeout;
+        $deadline = microtime(true) + $timeout;
+
+        if (!$this->hasMessage($timeout)) {
+            return null;
+        }
+
+        try {
+            $size = $socket->read(Bytes::BYTES_SIZE);
+
+            if ('' === $size) {
+                $this->disconnect();
+
+                throw new ConnectionFail('Probably connection lost');
+            }
+
+            if ($this->connectionConfig?->snappy) {
+                $buffer = new ByteBuffer();
+                $snappyBuffer = new ByteBuffer($size);
+                while (true) {
+                    $typeByte = \ord($snappyBuffer->consume(1));
+
+                    $size = \ord($snappyBuffer->consume(1)) + (\ord($snappyBuffer->consume(1)) << 8) + (\ord($snappyBuffer->consume(1)) << 16);
+                    $type = match ($typeByte) {
+                        0xff => 'identifier',
+                        0x00 => 'compressed',
+                        0x01 => 'uncompressed',
+                        0xfe => 'padding',
+                    };
+
+                    $this->logger->debug('Received snappy chunk: {type}, size: {size}', [
+                        'type' => $type,
+                        'size' => $size,
+                    ]);
+
+                    switch ($typeByte) {
+                        case 0xff: // 'identifier',
+                            $socket->read($size);
+                            $snappyBuffer->append($socket->read(4));
+
+                            continue 2;
+                        case 0x00: // 'compressed',
+                        case 0x01: // 'uncompressed',
+                            $uncompressed = $socket->read($size);
+
+                            $this->logger->debug('Received uncompressed bytes: {bytes}', ['bytes' => $uncompressed]);
+                            $buffer->append($uncompressed);
+                            $buffer->consume(4); // slice snappy prefix
+                            $buffer->consumeUint32(); // slice size
+
+                            break 2;
+                        case 0xfe:// 'padding',
+                    }
+                }
+            } else {
+                $this->logger->debug('Size bytes received: "{bytes}"', ['bytes' => $size]);
+
+                $buffer = new ByteBuffer($size);
+
+                $size = $buffer->consumeUint32();
+
+                do {
+                    $chunk = $socket->read($size);
 
-                if (null === $response && null !== ($chunk = yield $stream->read())) {
                     $buffer->append($chunk);
 
-                    continue;
+                    $size -= \strlen($chunk);
-                }
+                } while (0 < $size);
-
-                if (!$response instanceof Frame\Response) {
-                    throw new NsqException();
-                }
-
-                return $response;
             }
-        });
+
+            $this->logger->debug('Received buffer: '.addcslashes($buffer->bytes(), PHP_EOL));
+
+            $response = new Response($buffer);
+
+            if ($response->isHeartBeat()) {
+                $this->command('NOP');
+
+                return $this->receive(
+                    ($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime
+                );
+            }
+        }
+        // @codeCoverageIgnoreStart
+        catch (Exception $e) {
+            $this->disconnect();
+
+            throw ConnectionFail::fromThrowable($e);
+        }
+        // @codeCoverageIgnoreEnd
+
+        return $response;
+    }
+
+    protected function response(): Response
+    {
+        return $this->receive() ?? throw UnexpectedResponse::null();
+    }
+
+    private function socket(): Socket
+    {
+        if (null === $this->socket) {
+            $this->connect();
+        }
+
+        return $this->socket ?? throw new ConnectionFail('This connection is closed, create new one.');
     }
 }

src/Consumer.php

@@ -4,184 +4,60 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Amp\Failure;
-use Amp\Promise;
-use Nsq\Config\ClientConfig;
-use Nsq\Exception\ConsumerException;
-use Nsq\Frame\Response;
-use Nsq\Stream\NullStream;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
-use function Amp\asyncCall;
-use function Amp\call;
-
 final class Consumer extends Connection
 {
     private int $rdy = 0;
 
     /**
-     * @var callable
+     * Subscribe to a topic/channel.
      */
-    private $onMessage;
+    public function sub(string $topic, string $channel): void
-
-    public function __construct(
-        private string $address,
-        private string $topic,
-        private string $channel,
-        callable $onMessage,
-        ClientConfig $clientConfig,
-        private LoggerInterface $logger,
-    ) {
-        parent::__construct(
-            $this->address,
-            $clientConfig,
-            $this->logger,
-        );
-
-        $this->onMessage = $onMessage;
-    }
-
-    public static function create(
-        string $address,
-        string $topic,
-        string $channel,
-        callable $onMessage,
-        ?ClientConfig $clientConfig = null,
-        ?LoggerInterface $logger = null,
-    ): self {
-        return new self(
-            $address,
-            $topic,
-            $channel,
-            $onMessage,
-            $clientConfig ?? new ClientConfig(),
-            $logger ?? new NullLogger(),
-        );
-    }
-
-    public function connect(): Promise
     {
-        if (!$this->stream instanceof NullStream) {
+        $this->command('SUB', [$topic, $channel])->response()->okOrFail();
-            return call(static function (): void {
-            });
-        }
-
-        return call(function (): \Generator {
-            yield parent::connect();
-
-            $this->run();
-        });
-    }
-
-    private function run(): void
-    {
-        $buffer = new Buffer();
-
-        asyncCall(function () use ($buffer): \Generator {
-            yield $this->stream->write(Command::sub($this->topic, $this->channel));
-
-            if (null !== ($chunk = yield $this->stream->read())) {
-                $buffer->append($chunk);
-            }
-
-            /** @var Response $response */
-            $response = Parser::parse($buffer);
-
-            if (!$response->isOk()) {
-                return new Failure(new ConsumerException('Fail subscription.'));
-            }
-
-            yield $this->rdy(2500);
-
-            /** @phpstan-ignore-next-line  */
-            asyncCall(function () use ($buffer): \Generator {
-                while (null !== $chunk = yield $this->stream->read()) {
-                    $buffer->append($chunk);
-
-                    while ($frame = Parser::parse($buffer)) {
-                        switch (true) {
-                            case $frame instanceof Frame\Response:
-                                if ($frame->isHeartBeat()) {
-                                    yield $this->stream->write(Command::nop());
-
-                                    break;
-                                }
-
-                                throw ConsumerException::response($frame);
-                            case $frame instanceof Frame\Error:
-                                $this->handleError($frame);
-
-                                break;
-                            case $frame instanceof Frame\Message:
-                                asyncCall($this->onMessage, Message::compose($frame, $this));
-
-                                break;
-                        }
-                    }
-                }
-
-                $this->stream = new NullStream();
-            });
-        });
     }
 
     /**
      * Update RDY state (indicate you are ready to receive N messages).
-     *
-     * @return Promise<void>
      */
-    public function rdy(int $count): Promise
+    public function rdy(int $count): void
     {
         if ($this->rdy === $count) {
-            return call(static function (): void {
+            return;
-            });
         }
 
-        $this->rdy = $count;
+        $this->command('RDY', (string) $count);
 
-        return $this->stream->write(Command::rdy($count));
+        $this->rdy = $count;
     }
 
     /**
      * Finish a message (indicate successful processing).
-     *
-     * @return Promise<void>
-     *
-     * @internal
      */
-    public function fin(string $id): Promise
+    public function fin(string $id): void
     {
-        --$this->rdy;
+        $this->command('FIN', $id);
 
-        return $this->stream->write(Command::fin($id));
+        --$this->rdy;
     }
 
     /**
-     * Re-queue a message (indicate failure to process) The re-queued message is placed at the tail of the queue,
+     * Re-queue a message (indicate failure to process)
-     * equivalent to having just published it, but for various implementation specific reasons that behavior should not
+     * The re-queued message is placed at the tail of the queue, equivalent to having just published it,
-     * be explicitly relied upon and may change in the future. Similarly, a message that is in-flight and times out
+     * but for various implementation specific reasons that behavior should not be explicitly relied upon and may change in the future.
-     * behaves identically to an explicit REQ.
+     * Similarly, a message that is in-flight and times out behaves identically to an explicit REQ.
-     *
-     * @return Promise<void>
-     *
-     * @internal
      */
-    public function req(string $id, int $timeout): Promise
+    public function req(string $id, int $timeout): void
     {
-        --$this->rdy;
+        $this->command('REQ', [$id, $timeout]);
 
-        return $this->stream->write(Command::req($id, $timeout));
+        --$this->rdy;
     }
 
     /**
      * Reset the timeout for an in-flight message.
-     *
-     * @return Promise<void>
-     *
-     * @internal
      */
-    public function touch(string $id): Promise
+    public function touch(string $id): void
     {
-        return $this->stream->write(Command::touch($id));
+        $this->command('TOUCH', $id);
     }
 }

src/ErrorType.php

@@ -1,99 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-/**
- * @psalm-immutable
- */
-final class ErrorType
-{
-    /**
-     * A generic error type without any more hints.
-     */
-    public const E_INVALID = true;
-    /**
-     * This error might be returned during multiple occasions. It can be returned for IDENTIFY, AUTH or MPUB messages.
-     * It is caused for payloads that do not meet certain requirements. For IDENTIFY and AUTH, this is usually a bug in
-     * the library and should be reported. For MPUB, this error can occur if the payload is larger than the maximum
-     * payload size specified in the nsqd config.
-     */
-    public const E_BAD_BODY = true;
-    /**
-     * This error indicates that the topic sent to nsqd is not valid.
-     */
-    public const E_BAD_TOPIC = true;
-    /**
-     * This error indicates that the channel sent to nsqd is not valid.
-     */
-    public const E_BAD_CHANNEL = true;
-    /**
-     * This error is returned by nsqd if the message in the payload of a publishing operation does not meet the
-     * requirements of the server. This might be caused by too big payloads being sent to nsqd. You should consider
-     * adding a limit to the payload size or increasing it in the nsqd config.
-     */
-    public const E_BAD_MESSAGE = true;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
-     * temporary error and can be caused by topics being added, deleted or cleared.
-     */
-    public const E_PUB_FAILED = true;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
-     * temporary error and can be caused by topics being added, deleted or cleared.
-     */
-    public const E_MPUB_FAILED = true;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
-     * temporary error and can be caused by topics being added, deleted or cleared.
-     */
-    public const E_DPUB_FAILED = true;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This can
-     * happen in particular for messages that are no longer queued on the server side.
-     */
-    public const E_FIN_FAILED = false;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This can
-     * happen in particular for messages that are no longer queued on the server side.
-     */
-    public const E_REQ_FAILED = false;
-    /**
-     * This error may happen if a error condition is met after validating the input on the nsqd side. This can
-     * happen in particular for messages that are no longer queued on the server side.
-     */
-    public const E_TOUCH_FAILED = false;
-    /**
-     * This error indicates that the authorization of the client failed on the server side. This might be related
-     * to connection issues to the authorization server. Depending on the authorization server implementation, this
-     * might also indicate that the given auth secret in the [ClientConfig] is not known on the server or the server
-     * denied authentication with the current connection properties (i.e. TLS status and IP).
-     */
-    public const E_AUTH_FAILED = true;
-    /**
-     * This error happens if something breaks on the nsqd side while performing the authorization. This might be
-     * caused by bugs in nsqd, the authorization server or network issues.
-     */
-    public const E_AUTH_ERROR = true;
-    /**
-     * This error is sent by nsqd if the client attempts an authentication, but the server does not support it. This
-     * should never happen using this library as authorization requests are only sent if the server supports it.
-     * It is safe to expect that this error is never thrown.
-     */
-    public const E_AUTH_DISABLED = true;
-    /**
-     * This error indicates that the client related to the authorization secret set in the [ClientConfig] is not
-     * allowed to do the operation it tried to do.
-     */
-    public const E_UNAUTHORIZED = true;
-
-    public static function terminable(Frame\Error $error): bool
-    {
-        $type = explode(' ', $error->data)[0];
-
-        $constant = 'self::'.$type;
-
-        return \defined($constant) ? \constant($constant) : self::E_INVALID;
-    }
-}

src/Exception/AuthenticationRequired.php

@@ -4,10 +4,8 @@ declare(strict_types=1);
 
 namespace Nsq\Exception;
 
-final class AuthenticationRequired extends NsqException
+use RuntimeException;
+
+final class AuthenticationRequired extends RuntimeException implements NsqException
 {
-    public function __construct()
-    {
-        parent::__construct('NSQ requires authorization, set ClientConfig::$authSecret before connecting');
-    }
 }

src/Exception/ConnectionFail.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Exception;
+
+use RuntimeException;
+use Throwable;
+
+final class ConnectionFail extends RuntimeException implements NsqException
+{
+    /**
+     * @codeCoverageIgnore
+     */
+    public static function fromThrowable(Throwable $throwable): self
+    {
+        return new self($throwable->getMessage(), (int) $throwable->getCode(), $throwable);
+    }
+}

src/Exception/ConsumerException.php

@@ -1,15 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-use Nsq\Frame\Response;
-
-final class ConsumerException extends NsqException
-{
-    public static function response(Response $response): self
-    {
-        return new self(sprintf('Consumer receive response "%s" from nsq, which not expected. ', $response->data));
-    }
-}

src/Exception/MessageAlreadyFinished.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Exception;
+
+use Nsq\Message;
+use RuntimeException;
+
+final class MessageAlreadyFinished extends RuntimeException implements NsqException
+{
+    public static function finish(Message $message): self
+    {
+        return new self('Can\'t finish message as it already finished.');
+    }
+
+    public static function requeue(Message $message): self
+    {
+        return new self('Can\'t requeue message as it already finished.');
+    }
+
+    public static function touch(Message $message): self
+    {
+        return new self('Can\'t touch message as it already finished.');
+    }
+}

src/Exception/MessageException.php

@@ -1,15 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-use Nsq\Message;
-
-final class MessageException extends NsqException
-{
-    public static function processed(Message $message): self
-    {
-        return new self(sprintf('Message "%s" already processed.', $message->id));
-    }
-}

src/Exception/NsqError.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Exception;
+
+use RuntimeException;
+
+final class NsqError extends RuntimeException implements NsqException
+{
+}

src/Exception/NsqException.php

@@ -4,6 +4,8 @@ declare(strict_types=1);
 
 namespace Nsq\Exception;
 
-class NsqException extends \RuntimeException
+use Throwable;
+
+interface NsqException extends Throwable
 {
 }

src/Exception/ServerException.php

@@ -1,9 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-final class ServerException extends NsqException
-{
-}

src/Exception/SnappyException.php

@@ -1,18 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-final class SnappyException extends NsqException
-{
-    public static function notInstalled(): self
-    {
-        return new self('Snappy extension not installed.');
-    }
-
-    public static function invalidHeader(): self
-    {
-        return new self('Invalid snappy protocol header.');
-    }
-}

src/Exception/UnexpectedResponse.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Exception;
+
+use RuntimeException;
+
+final class UnexpectedResponse extends RuntimeException implements NsqException
+{
+    /**
+     * @codeCoverageIgnore
+     */
+    public static function null(): self
+    {
+        return new self('Response was expected, but null received.');
+    }
+}

src/Frame.php

@@ -1,32 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-abstract class Frame
-{
-    public const TYPE_RESPONSE = 0,
-        TYPE_ERROR = 1,
-        TYPE_MESSAGE = 2
-    ;
-
-    public function __construct(public int $type)
-    {
-    }
-
-    public function response(): bool
-    {
-        return self::TYPE_RESPONSE === $this->type;
-    }
-
-    public function error(): bool
-    {
-        return self::TYPE_ERROR === $this->type;
-    }
-
-    public function message(): bool
-    {
-        return self::TYPE_MESSAGE === $this->type;
-    }
-}

src/Frame/Error.php

@@ -1,24 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Frame;
-
-use Nsq\Exception\ServerException;
-use Nsq\Frame;
-
-/**
- * @psalm-immutable
- */
-final class Error extends Frame
-{
-    public function __construct(public string $data)
-    {
-        parent::__construct(self::TYPE_ERROR);
-    }
-
-    public function toException(): ServerException
-    {
-        return new ServerException($this->data);
-    }
-}

src/Frame/Message.php

@@ -1,19 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Frame;
-
-use Nsq\Frame;
-
-final class Message extends Frame
-{
-    public function __construct(
-        public int $timestamp,
-        public int $attempts,
-        public string $id,
-        public string $body,
-    ) {
-        parent::__construct(self::TYPE_MESSAGE);
-    }
-}

src/Frame/Response.php

@@ -1,39 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Frame;
-
-use Nsq\Frame;
-
-/**
- * @psalm-immutable
- */
-final class Response extends Frame
-{
-    public const OK = 'OK';
-    public const HEARTBEAT = '_heartbeat_';
-
-    public function __construct(public string $data)
-    {
-        parent::__construct(self::TYPE_RESPONSE);
-    }
-
-    public function isOk(): bool
-    {
-        return self::OK === $this->data;
-    }
-
-    public function isHeartBeat(): bool
-    {
-        return self::HEARTBEAT === $this->data;
-    }
-
-    /**
-     * @return array<mixed, mixed>
-     */
-    public function toArray(): array
-    {
-        return json_decode($this->data, true, flags: JSON_THROW_ON_ERROR);
-    }
-}

src/Message.php

@@ -4,79 +4,75 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Amp\Promise;
+use Nsq\Exception\MessageAlreadyFinished;
-use Nsq\Exception\MessageException;
-use function Amp\call;
 
 final class Message
 {
-    private bool $processed = false;
+    /**
-
+     * @psalm-readonly
-    public function __construct(
+     */
-        public string $id,
+    public int $timestamp;
-        public string $body,
-        public int $timestamp,
-        public int $attempts,
-        private Consumer $consumer,
-    ) {
-    }
-
-    public static function compose(Frame\Message $message, Consumer $consumer): self
-    {
-        return new self(
-            $message->id,
-            $message->body,
-            $message->timestamp,
-            $message->attempts,
-            $consumer,
-        );
-    }
 
     /**
-     * @return Promise<void>
+     * @psalm-readonly
      */
-    public function finish(): Promise
+    public int $attempts;
-    {
-        return call(function (): \Generator {
-            if ($this->processed) {
-                throw MessageException::processed($this);
-            }
-
-            yield $this->consumer->fin($this->id);
-
-            $this->processed = true;
-        });
-    }
 
     /**
-     * @return Promise<void>
+     * @psalm-readonly
      */
-    public function requeue(int $timeout): Promise
+    public string $id;
-    {
-        return call(function () use ($timeout): \Generator {
-            if ($this->processed) {
-                throw MessageException::processed($this);
-            }
-
-            yield $this->consumer->req($this->id, $timeout);
-
-            $this->processed = true;
-        });
-    }
 
     /**
-     * @return Promise<void>
+     * @psalm-readonly
      */
-    public function touch(): Promise
+    public string $body;
+
+    private bool $finished = false;
+
+    private Consumer $consumer;
+
+    public function __construct(int $timestamp, int $attempts, string $id, string $body, Consumer $consumer)
     {
-        return call(function (): \Generator {
+        $this->timestamp = $timestamp;
-            if ($this->processed) {
+        $this->attempts = $attempts;
-                throw MessageException::processed($this);
+        $this->id = $id;
-            }
+        $this->body = $body;
 
-            yield $this->consumer->touch($this->id);
+        $this->consumer = $consumer;
+    }
 
-            $this->processed = true;
+    public function isFinished(): bool
-        });
+    {
+        return $this->finished;
+    }
+
+    public function finish(): void
+    {
+        if ($this->finished) {
+            throw MessageAlreadyFinished::finish($this);
+        }
+
+        $this->consumer->fin($this->id);
+        $this->finished = true;
+    }
+
+    public function requeue(int $timeout): void
+    {
+        if ($this->finished) {
+            throw MessageAlreadyFinished::requeue($this);
+        }
+
+        $this->consumer->req($this->id, $timeout);
+        $this->finished = true;
+    }
+
+    public function touch(): void
+    {
+        if ($this->finished) {
+            throw MessageAlreadyFinished::touch($this);
+        }
+
+        $this->consumer->touch($this->id);
     }
 }

src/Parser.php

@@ -1,47 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-use Nsq\Exception\NsqException;
-
-class Parser
-{
-    private const SIZE = 4;
-    private const TYPE = 4;
-    private const MESSAGE_HEADER_SIZE =
-        8 +  // timestamp
-        2 +  // attempts
-        16 + // ID
-        4;   // Frame type
-
-    public static function parse(Buffer $buffer): ?Frame
-    {
-        if ($buffer->size() < self::SIZE) {
-            return null;
-        }
-
-        $size = $buffer->readInt32();
-
-        if ($buffer->size() < $size + self::SIZE) {
-            return null;
-        }
-
-        $buffer->discard(self::SIZE);
-
-        $type = $buffer->consumeInt32();
-
-        return match ($type) {
-            Frame::TYPE_RESPONSE => new Frame\Response($buffer->consume($size - self::TYPE)),
-            Frame::TYPE_ERROR => new Frame\Error($buffer->consume($size - self::TYPE)),
-            Frame::TYPE_MESSAGE => new Frame\Message(
-                timestamp: $buffer->consumeTimestamp(),
-                attempts: $buffer->consumeAttempts(),
-                id: $buffer->consumeMessageID(),
-                body: $buffer->consume($size - self::MESSAGE_HEADER_SIZE),
-            ),
-            default => throw new NsqException(sprintf('Unexpected frame type: "%s"', $type)),
-        };
-    }
-}

src/Producer.php

@@ -4,96 +4,41 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Amp\Promise;
+use function array_map;
-use Nsq\Config\ClientConfig;
+use function implode;
-use Nsq\Exception\NsqException;
+use function pack;
-use Nsq\Stream\NullStream;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
-use function Amp\asyncCall;
-use function Amp\call;
 
 final class Producer extends Connection
 {
-    public static function create(
+    /**
-        string $address,
+     * @psalm-suppress PossiblyFalseOperand
-        ClientConfig $clientConfig = null,
+     */
-        LoggerInterface $logger = null,
+    public function pub(string $topic, string $body): void
-    ): self {
-        return new self(
-            $address,
-            $clientConfig ?? new ClientConfig(),
-            $logger ?? new NullLogger(),
-        );
-    }
-
-    public function connect(): Promise
     {
-        if (!$this->stream instanceof NullStream) {
+        $this->command('PUB', $topic, $body)->response()->okOrFail();
-            return call(static function (): void {
-            });
-        }
-
-        return call(function (): \Generator {
-            yield parent::connect();
-
-            $this->run();
-        });
     }
 
     /**
-     * @param array<int, string>|string $body
+     * @psalm-param array<mixed, mixed> $bodies
      *
-     * @return Promise<void>
+     * @psalm-suppress PossiblyFalseOperand
      */
-    public function publish(string $topic, string | array $body, int $delay = 0): Promise
+    public function mpub(string $topic, array $bodies): void
     {
-        if (0 < $delay) {
+        $num = pack('N', \count($bodies));
-            return call(
-                function (array $bodies) use ($topic, $delay): \Generator {
-                    foreach ($bodies as $body) {
-                        yield $this->stream->write(Command::dpub($topic, $body, $delay));
-                    }
-                },
-                (array) $body,
-            );
-        }
 
-        $command = \is_array($body)
+        $mb = implode('', array_map(static function ($body): string {
-            ? Command::mpub($topic, $body)
+            return pack('N', \strlen($body)).$body;
-            : Command::pub($topic, $body);
+        }, $bodies));
 
-        return $this->stream->write($command);
+        $this->command('MPUB', $topic, $num.$mb)->response()->okOrFail();
     }
 
-    private function run(): void
+    /**
+     * @psalm-suppress PossiblyFalseOperand
+     */
+    public function dpub(string $topic, string $body, int $delay): void
     {
-        $buffer = new Buffer();
+        $this->command('DPUB', [$topic, $delay], $body)->response()->okOrFail();
-
-        asyncCall(function () use ($buffer): \Generator {
-            while (null !== $chunk = yield $this->stream->read()) {
-                $buffer->append($chunk);
-
-                while ($frame = Parser::parse($buffer)) {
-                    switch (true) {
-                        case $frame instanceof Frame\Response:
-                            if ($frame->isHeartBeat()) {
-                                yield $this->stream->write(Command::nop());
-                            }
-
-                            // Ok received
-                            break;
-                        case $frame instanceof Frame\Error:
-                            $this->handleError($frame);
-
-                            break;
-                        default:
-                            throw new NsqException('Unreachable statement.');
-                    }
-                }
-            }
-
-            $this->stream = new NullStream();
-        });
     }
 }

src/Reconnect/ExponentialStrategy.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Reconnect;
+
+use Nsq\Exception\ConnectionFail;
+use Psr\Log\LoggerAwareTrait;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Throwable;
+
+final class ExponentialStrategy implements ReconnectStrategy
+{
+    use LoggerAwareTrait;
+
+    private int $delay;
+
+    private int $nextTryAfter;
+
+    private int $attempt = 0;
+
+    private TimeProvider $timeProvider;
+
+    public function __construct(
+        private int $minDelay = 8,
+        private int $maxDelay = 32,
+        TimeProvider $timeProvider = null,
+        LoggerInterface $logger = null,
+    ) {
+        $this->delay = 0;
+        $this->timeProvider = $timeProvider ?? new RealTimeProvider();
+        $this->nextTryAfter = $this->timeProvider->time();
+        $this->logger = $logger ?? new NullLogger();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function connect(callable $callable): void
+    {
+        $currentTime = $this->timeProvider->time();
+
+        if ($currentTime < $this->nextTryAfter) {
+            throw new ConnectionFail('Time to reconnect has not yet come');
+        }
+
+        try {
+            $callable();
+        } catch (Throwable $e) {
+            $nextDelay = 0 === $this->delay ? $this->minDelay : $this->delay * 2;
+            $this->delay = $nextDelay > $this->maxDelay ? $this->maxDelay : $nextDelay;
+            $this->nextTryAfter = $currentTime + $this->delay;
+
+            $this->logger->warning('Reconnect #{attempt} after {delay}s', ['attempt' => ++$this->attempt, 'delay' => $this->delay]);
+
+            throw $e;
+        }
+
+        $this->delay = 0;
+        $this->attempt = 0;
+    }
+}

src/Reconnect/RealTimeProvider.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Reconnect;
+
+final class RealTimeProvider implements TimeProvider
+{
+    public function time(): int
+    {
+        return time();
+    }
+}

src/Reconnect/ReconnectStrategy.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Reconnect;
+
+use Nsq\Exception\ConnectionFail;
+
+interface ReconnectStrategy
+{
+    /**
+     * @throws ConnectionFail
+     */
+    public function connect(callable $callable): void;
+}

src/Reconnect/TimeProvider.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\Reconnect;
+
+interface TimeProvider
+{
+    public function time(): int;
+}

src/Response.php

@@ -0,0 +1,87 @@
+<?php
+
+declare(strict_types=1);
+
+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
+{
+    private const OK = 'OK';
+    private const HEARTBEAT = '_heartbeat_';
+    private const TYPE_RESPONSE = 0;
+    private const TYPE_ERROR = 1;
+    private const TYPE_MESSAGE = 2;
+
+    private int $type;
+
+    private ByteBuffer $buffer;
+
+    public function __construct(ByteBuffer $buffer)
+    {
+        $this->type = $buffer->consumeUint32();
+        $this->buffer = $buffer;
+    }
+
+    public function okOrFail(): void
+    {
+        if (self::TYPE_ERROR === $this->type) {
+            throw new NsqError($this->buffer->bytes());
+        }
+
+        if (self::TYPE_RESPONSE !== $this->type) {
+            // @codeCoverageIgnoreStart
+            throw new UnexpectedResponse(sprintf('"%s" type expected, but "%s" received.', self::TYPE_RESPONSE, $this->type));
+            // @codeCoverageIgnoreEnd
+        }
+
+        if (self::OK !== $this->buffer->bytes()) {
+            // @codeCoverageIgnoreStart
+            throw new UnexpectedResponse(sprintf('OK response expected, but "%s" received.', $this->buffer->bytes()));
+            // @codeCoverageIgnoreEnd
+        }
+    }
+
+    public function isHeartBeat(): bool
+    {
+        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) {
+            // @codeCoverageIgnoreStart
+            throw new UnexpectedResponse(sprintf('Expecting "%s" type, but NSQ return: "%s"', self::TYPE_MESSAGE, $this->type));
+            // @codeCoverageIgnoreEnd
+        }
+
+        $buffer = new ByteBuffer($this->buffer->bytes());
+
+        $timestamp = $buffer->consumeInt64();
+        $attempts = $buffer->consumeUint16();
+        $id = $buffer->consume(Bytes::BYTES_ID);
+        $body = $buffer->flush();
+
+        return new Message($timestamp, $attempts, $id, $body, $reader);
+    }
+}

src/Stream.php

@@ -1,22 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-use Amp\Promise;
-
-interface Stream
-{
-    /**
-     * @return Promise<null|string>
-     */
-    public function read(): Promise;
-
-    /**
-     * @return Promise<void>
-     */
-    public function write(string $data): Promise;
-
-    public function close(): void;
-}

src/Stream/GzipStream.php

@@ -1,38 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\Promise;
-use Nsq\Exception\NsqException;
-use Nsq\Stream;
-
-class GzipStream implements Stream
-{
-    public function __construct(private Stream $stream)
-    {
-        throw new NsqException('GzipStream not implemented yet.');
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function read(): Promise
-    {
-        return $this->stream->read();
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function write(string $data): Promise
-    {
-        return $this->stream->write($data);
-    }
-
-    public function close(): void
-    {
-        $this->stream->close();
-    }
-}

src/Stream/NullStream.php

@@ -1,37 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\Promise;
-use Amp\Success;
-use Nsq\Stream;
-use function Amp\call;
-
-final class NullStream implements Stream
-{
-    /**
-     * {@inheritdoc}
-     */
-    public function read(): Promise
-    {
-        return new Success(null);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function write(string $data): Promise
-    {
-        return call(static function (): void {
-        });
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function close(): void
-    {
-    }
-}

src/Stream/SnappyStream.php

@@ -1,115 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\Promise;
-use Nsq\Buffer;
-use Nsq\Exception\SnappyException;
-use Nsq\Stream;
-use function Amp\call;
-
-class SnappyStream implements Stream
-{
-    private const IDENTIFIER = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
-    private const SIZE_HEADER = 4;
-    private const SIZE_CHECKSUM = 4;
-    private const SIZE_CHUNK = 65536;
-    private const TYPE_IDENTIFIER = 0xff;
-    private const TYPE_COMPRESSED = 0x00;
-    private const TYPE_UNCOMPRESSED = 0x01;
-    private const TYPE_PADDING = 0xfe;
-
-    private Buffer $buffer;
-
-    public function __construct(private Stream $stream, string $bytes = '')
-    {
-        if (!\function_exists('snappy_uncompress')) {
-            throw SnappyException::notInstalled();
-        }
-
-        $this->buffer = new Buffer($bytes);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function read(): Promise
-    {
-        return call(function (): \Generator {
-            if ($this->buffer->size() < self::SIZE_HEADER && null !== ($chunk = yield $this->stream->read())) {
-                $this->buffer->append($chunk);
-            }
-
-            $type = $this->buffer->readUInt32LE();
-
-            $size = $type >> 8;
-            $type &= 0xff;
-
-            while ($this->buffer->size() < $size && null !== ($chunk = yield $this->stream->read())) {
-                $this->buffer->append($chunk);
-            }
-
-            switch ($type) {
-                case self::TYPE_IDENTIFIER:
-                    $this->buffer->discard($size);
-
-                    return $this->read();
-                case self::TYPE_COMPRESSED:
-                    $this->buffer->discard(self::SIZE_CHECKSUM);
-
-                    return snappy_uncompress($this->buffer->consume($size - self::SIZE_HEADER));
-                case self::TYPE_UNCOMPRESSED:
-                    $this->buffer->discard(self::SIZE_CHECKSUM);
-
-                    return $this->buffer->consume($size - self::SIZE_HEADER);
-                case self::TYPE_PADDING:
-                    return $this->read();
-                default:
-                    throw SnappyException::invalidHeader();
-            }
-        });
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function write(string $data): Promise
-    {
-        return call(function () use ($data): Promise {
-            $result = pack('CCCCCCCCCC', ...self::IDENTIFIER);
-
-            foreach (str_split($data, self::SIZE_CHUNK) as $chunk) {
-                $result .= $this->compress($chunk);
-            }
-
-            return $this->stream->write($result);
-        });
-    }
-
-    public function close(): void
-    {
-        $this->stream->close();
-    }
-
-    /**
-     * @psalm-suppress PossiblyFalseArgument
-     */
-    private function compress(string $uncompressed): string
-    {
-        $compressed = snappy_compress($uncompressed);
-
-        [$type, $data] = \strlen($compressed) <= 0.875 * \strlen($uncompressed)
-            ? [self::TYPE_COMPRESSED, $compressed]
-            : [self::TYPE_UNCOMPRESSED, $uncompressed];
-
-        /** @phpstan-ignore-next-line */
-        $checksum = unpack('N', hash('crc32c', $uncompressed, true))[1];
-        $checksum = (($checksum >> 15) | ($checksum << 17)) + 0xa282ead8 & 0xffffffff;
-
-        $size = (\strlen($data) + 4) << 8;
-
-        return pack('VV', $type + $size, $checksum).$data;
-    }
-}

src/Stream/SocketStream.php

@@ -1,64 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\Promise;
-use Amp\Socket\ConnectContext;
-use Amp\Socket\Socket;
-use Nsq\Stream;
-use function Amp\call;
-use function Amp\Socket\connect;
-
-class SocketStream implements Stream
-{
-    public function __construct(private Socket $socket)
-    {
-    }
-
-    /**
-     * @return Promise<self>
-     */
-    public static function connect(string $uri, int $timeout = 0, int $attempts = 0, bool $noDelay = false): Promise
-    {
-        return call(function () use ($uri, $timeout, $attempts, $noDelay): \Generator {
-            $context = new ConnectContext();
-
-            if ($timeout > 0) {
-                $context = $context->withConnectTimeout($timeout);
-            }
-
-            if ($attempts > 0) {
-                $context = $context->withMaxAttempts($attempts);
-            }
-
-            if ($noDelay) {
-                $context = $context->withTcpNoDelay();
-            }
-
-            return new self(yield connect($uri, $context));
-        });
-    }
-
-    /**
-     * @return Promise<null|string>
-     */
-    public function read(): Promise
-    {
-        return $this->socket->read();
-    }
-
-    /**
-     * @return Promise<void>
-     */
-    public function write(string $data): Promise
-    {
-        return $this->socket->write($data);
-    }
-
-    public function close(): void
-    {
-        $this->socket->close();
-    }
-}

src/Subscriber.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq;
+
+use Generator;
+
+final class Subscriber
+{
+    public const STOP = 0;
+
+    private Consumer $reader;
+
+    public function __construct(Consumer $reader)
+    {
+        $this->reader = $reader;
+    }
+
+    /**
+     * @psalm-return Generator<int, Message|float|null, int|float|null, void>
+     */
+    public function subscribe(string $topic, string $channel): Generator
+    {
+        $this->reader->sub($topic, $channel);
+
+        while (true) {
+            $this->reader->rdy(1);
+
+            $command = yield $this->reader->receive()?->toMessage($this->reader);
+
+            if (self::STOP === $command) {
+                break;
+            }
+        }
+
+        $this->reader->disconnect();
+    }
+}

tests/ErrorTypeTest.php

@@ -1,28 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-use Nsq\ErrorType;
-use Nsq\Frame\Error;
-use PHPUnit\Framework\TestCase;
-
-final class ErrorTypeTest extends TestCase
-{
-    /**
-     * @dataProvider data
-     */
-    public function testConstructor(Error $frame, bool $isConnectionTerminated): void
-    {
-        self::assertSame($isConnectionTerminated, ErrorType::terminable($frame));
-    }
-
-    /**
-     * @return \Generator<int, array{0: Error, 1: bool}>
-     */
-    public function data(): Generator
-    {
-        yield [new Error('E_BAD_BODY'), true];
-        yield [new Error('bla_bla'), true];
-        yield [new Error('E_REQ_FAILED'), false];
-    }
-}

tests/ExponentialStrategyTest.php

@@ -0,0 +1,90 @@
+<?php
+
+declare(strict_types=1);
+
+use Nsq\Exception\ConnectionFail;
+use Nsq\Reconnect\ExponentialStrategy;
+use Nsq\Reconnect\TimeProvider;
+use PHPUnit\Framework\TestCase;
+
+final class ExponentialStrategyTest extends TestCase
+{
+    public function testTimeNotYetCome(): void
+    {
+        $timeProvider = new FakeTimeProvider();
+        $strategy = new ExponentialStrategy(
+            minDelay: 8,
+            maxDelay: 32,
+            timeProvider: $timeProvider,
+        );
+
+        $successConnect = static function (int $time = null) use ($strategy, $timeProvider): void {
+            $timeProvider($time);
+
+            $strategy->connect(static function (): void {
+            });
+        };
+        $failConnect = static function (int $time = null) use ($strategy, $timeProvider): void {
+            $timeProvider($time);
+
+            try {
+                $strategy->connect(function (): void {
+                    throw new ConnectionFail('Time come but failed');
+                });
+            } catch (ConnectionFail $e) {
+                self::assertSame('Time come but failed', $e->getMessage());
+
+                return;
+            }
+
+            self::fail('Expecting exception with message "Time come but failed"');
+        };
+        $timeNotCome = static function (int $time = null) use ($strategy, $timeProvider): void {
+            $timeProvider($time);
+
+            try {
+                $strategy->connect(function (): void {
+                    throw new ConnectionFail('');
+                });
+            } catch (ConnectionFail $e) {
+                self::assertSame('Time to reconnect has not yet come', $e->getMessage());
+
+                return;
+            }
+
+            self::fail('Was expecting exception with message "Time to reconnect has not yet come"');
+        };
+
+        $failConnect(0);
+        $timeNotCome(7);
+        $failConnect(8);
+        $timeNotCome(22);
+        $timeNotCome(13);
+        $failConnect(24);
+        $successConnect(56);
+        $failConnect();
+        $timeNotCome();
+        $timeNotCome(63);
+        $failConnect(64);
+
+        $this->expectException(ConnectionFail::class);
+        $this->expectExceptionMessage('Time to reconnect has not yet come');
+
+        $successConnect();
+    }
+}
+
+class FakeTimeProvider implements TimeProvider
+{
+    public int $time = 0;
+
+    public function time(): int
+    {
+        return $this->time;
+    }
+
+    public function __invoke(int $time = null): void
+    {
+        $this->time = $time ?? $this->time;
+    }
+}

tests/MessageTest.php

@@ -2,10 +2,8 @@
 
 declare(strict_types=1);
 
-use Amp\Loop;
-use Amp\Success;
 use Nsq\Consumer;
-use Nsq\Exception\MessageException;
+use Nsq\Exception\MessageAlreadyFinished;
 use Nsq\Message;
 use PHPUnit\Framework\TestCase;
 
@@ -16,12 +14,16 @@ final class MessageTest extends TestCase
      */
     public function testDoubleFinish(Message $message): void
     {
-        $this->expectException(MessageException::class);
+        self::assertFalse($message->isFinished());
 
-        Loop::run(function () use ($message): Generator {
+        $message->finish();
-            yield $message->finish();
+
-            yield $message->finish();
+        self::assertTrue($message->isFinished());
-        });
+
+        $this->expectException(MessageAlreadyFinished::class);
+        $this->expectExceptionMessage('Can\'t finish message as it already finished.');
+
+        $message->finish();
     }
 
     /**
@@ -29,12 +31,16 @@ final class MessageTest extends TestCase
      */
     public function testDoubleRequeue(Message $message): void
     {
-        $this->expectException(MessageException::class);
+        self::assertFalse($message->isFinished());
 
-        Loop::run(function () use ($message): Generator {
+        $message->requeue(1);
-            yield $message->requeue(1);
+
-            yield $message->requeue(5);
+        self::assertTrue($message->isFinished());
-        });
+
+        $this->expectException(MessageAlreadyFinished::class);
+        $this->expectExceptionMessage('Can\'t requeue message as it already finished.');
+
+        $message->requeue(5);
     }
 
     /**
@@ -42,12 +48,14 @@ final class MessageTest extends TestCase
      */
     public function testTouchAfterFinish(Message $message): void
     {
-        $this->expectException(MessageException::class);
+        self::assertFalse($message->isFinished());
 
-        Loop::run(function () use ($message): Generator {
+        $message->finish();
-            yield $message->finish();
+
-            yield $message->touch();
+        $this->expectException(MessageAlreadyFinished::class);
-        });
+        $this->expectExceptionMessage('Can\'t touch message as it already finished.');
+
+        $message->touch();
     }
 
     /**
@@ -55,11 +63,6 @@ final class MessageTest extends TestCase
      */
     public function messages(): Generator
     {
-        $consumer = $this->createMock(Consumer::class);
+        yield [new Message(0, 0, 'id', 'body', $this->createStub(Consumer::class))];
-        $consumer->method('fin')->willReturn(new Success());
-        $consumer->method('touch')->willReturn(new Success());
-        $consumer->method('req')->willReturn(new Success());
-
-        yield [new Message('id', 'body', 0, 0, $consumer)];
     }
 }

tests/NsqTest.php

@@ -0,0 +1,92 @@
+<?php
+
+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
+{
+    public function test(): void
+    {
+        $producer = new Producer('tcp://localhost:4150');
+        $producer->pub(__FUNCTION__, __FUNCTION__);
+
+        $consumer = new Consumer(
+            address: 'tcp://localhost:4150',
+            clientConfig: new ClientConfig(
+                heartbeatInterval: 1000,
+                readTimeout: 1,
+            ),
+        );
+        $subscriber = new Subscriber($consumer);
+        $generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__);
+
+        /** @var null|Message $message */
+        $message = $generator->current();
+
+        self::assertInstanceOf(Message::class, $message);
+        self::assertSame(__FUNCTION__, $message->body);
+        $message->finish();
+
+        $generator->next();
+        self::assertNull($generator->current());
+
+        $producer->mpub(__FUNCTION__, [
+            'First mpub message.',
+            'Second mpub message.',
+        ]);
+
+        $generator->next();
+        /** @var null|Message $message */
+        $message = $generator->current();
+        self::assertInstanceOf(Message::class, $message);
+        self::assertSame('First mpub message.', $message->body);
+        $message->finish();
+
+        $generator->next();
+        /** @var null|Message $message */
+        $message = $generator->current();
+        self::assertInstanceOf(Message::class, $message);
+        self::assertSame('Second mpub message.', $message->body);
+        $message->requeue(0);
+
+        $generator->next();
+        /** @var null|Message $message */
+        $message = $generator->current();
+        self::assertInstanceOf(Message::class, $message);
+        self::assertSame('Second mpub message.', $message->body);
+        $message->finish();
+
+        $producer->dpub(__FUNCTION__, 'Deferred message.', 2000);
+
+        $generator->next();
+        /** @var null|Message $message */
+        $message = $generator->current();
+        self::assertNull($message);
+
+        NSA::setProperty(
+            NSA::getProperty($consumer, 'clientConfig'),
+            'readTimeout',
+            10,
+        );
+
+        $generator->next();
+
+        /** @var null|Message $message */
+        $message = $generator->current();
+        self::assertInstanceOf(Message::class, $message);
+        self::assertSame('Deferred message.', $message->body);
+        $message->touch();
+        $message->finish();
+
+        self::assertTrue($consumer->isReady());
+        $generator->send(Subscriber::STOP);
+        self::assertFalse($consumer->isReady());
+    }
+}

tests/ProducerTest.php

@@ -2,60 +2,22 @@
 
 declare(strict_types=1);
 
-use Amp\Loop;
+use Nsq\Exception\NsqError;
-use Amp\Process\Process;
-use Nsq\Exception\ServerException;
 use Nsq\Producer;
 use PHPUnit\Framework\TestCase;
-use function Amp\ByteStream\buffer;
-use function Amp\Promise\wait;
 
 final class ProducerTest extends TestCase
 {
-    /**
-     * @param array<int, string>|string $body
-     *
-     * @dataProvider data
-     */
-    public function testPublish(array | string $body, string $expected): void
-    {
-        $process = new Process(
-            sprintf('bin/nsq_tail -topic %s -channel default -nsqd-tcp-address localhost:4150 -n 1', __FUNCTION__),
-        );
-        wait($process->start());
-
-        $producer = Producer::create('tcp://localhost:4150');
-        wait($producer->connect());
-        wait($producer->publish(__FUNCTION__, $body));
-
-        wait($process->join());
-
-        self::assertSame($expected, wait(buffer($process->getStdout())));
-    }
-
-    /**
-     * @return Generator<int, array{0: string|array, 1: string}>
-     */
-    public function data(): Generator
-    {
-        yield ['Test Message One!', 'Test Message One!'.PHP_EOL];
-    }
-
     /**
      * @dataProvider pubFails
      */
     public function testPubFail(string $topic, string $body, string $exceptionMessage): void
     {
-        $this->expectException(ServerException::class);
+        $this->expectException(NsqError::class);
         $this->expectExceptionMessage($exceptionMessage);
 
-        $producer = Producer::create('tcp://localhost:4150');
+        $producer = new Producer('tcp://localhost:4150');
-
+        $producer->pub($topic, $body);
-        Loop::run(static function () use ($producer, $topic, $body): Generator {
-            yield $producer->connect();
-
-            yield $producer->publish($topic, $body);
-        });
     }
 
     /**