Remove Producer::defer

.gitattributes

@@ -0,0 +1,21 @@
+# 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,6 +33,7 @@ jobs:
                 with:
                     php-version: ${{ matrix.php }}
                     coverage: pcov
+                    extensions: kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -108,6 +109,7 @@ jobs:
                 uses: shivammathur/setup-php@v2
                 with:
                     php-version: '8.0'
+                    extensions: snappy-kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -139,6 +141,7 @@ jobs:
                 uses: shivammathur/setup-php@v2
                 with:
                     php-version: '8.0'
+                    extensions: snappy-kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 

.php_cs.dist

@@ -8,6 +8,9 @@ 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'],
         ],

README.md

@@ -1,6 +1,6 @@
 # Nsq PHP
 
-<img src="https://github.com/nsqphp/nsqphp/raw/main/logo.png" alt="" align="left" width="150">
+<img src="https://github.com/nsqphp/nsqphp/raw/main/docs/logo.png" alt="" align="left" width="150">
 
 PHP Client for [NSQ](https://nsq.io/).
 
@@ -47,53 +47,37 @@ Usage
 ```php
 use Nsq\Producer;
 
-$producer = new Producer(address: 'tcp://nsqd:4150');
+$producer = Producer::create(address: 'tcp://nsqd:4150');
 
 // Publish a message to a topic
-$producer->pub('topic', 'Simple message');
+$producer->publish('topic', 'Simple message');
 
 // Publish multiple messages to a topic (atomically) 
-$producer->mpub('topic', [
+$producer->publish('topic', [
     'Message one',
     'Message two',
 ]);
 
 // Publish a deferred message to a topic
-$producer->dpub('topic', 'Deferred message', delay: 5000);
+$producer->publish('topic', 'Deferred message', delay: 5000);
 ```
 
 ### Consumer
 
 ```php
 use Nsq\Consumer;
-use Nsq\Protocol\Message;
+use Nsq\Message;
 
-$consumer = new Consumer(
+$consumer = Consumer::create(
+    address: 'tcp://nsqd:4150', 
     topic: 'topic',
     channel: 'channel',
-    address: 'tcp://nsqd:4150',
+    onMessage: static function (Message $message): Generator {
+        yield $message->touch(); // Reset the timeout for an in-flight message        
+        yield $message->requeue(timeout: 5000); // Re-queue a message (indicate failure to process)        
+        yield $message->finish(); // Finish a message (indicate successful processing)        
+    },
 );
-
-// Simple blocking loop based on generator
-$generator = $consumer->generator(); 
-
-foreach ($generator as $message) {
-    if ($message instanceof Message) {
-        $payload = $message->body;
-
-        // 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()
-
-    // Gracefully close connection (loop will be ended)
-    $generator->send(0); 
-}
 ```
 
 ### Integrations

composer.json

@@ -13,12 +13,13 @@
     "require": {
         "php": "^8.0.1",
         "ext-json": "*",
-        "clue/socket-raw": "^1.5",
+        "amphp/socket": "^1.1",
         "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",
@@ -42,7 +43,7 @@
     "prefer-stable": true,
     "scripts": {
         "cs": [
-            "vendor/bin/php-cs-fixer fix"
+            "vendor/bin/php-cs-fixer fix --using-cache=no"
         ],
         "cs-check": [
             "vendor/bin/php-cs-fixer fix --verbose --diff --dry-run"
@@ -59,7 +60,7 @@
             "vendor/bin/psalm"
         ],
         "test": [
-            "@norm-check",
+            "@norm",
             "@cs",
             "@phpstan",
             "@psalm",

docker-compose.yml

@@ -5,7 +5,8 @@ services:
         image: nsqio/nsq:v1.2.0
         labels:
             ru.grachevko.dhu: 'nsqd'
-        command: /nsqd
+        command: /nsqd -log-level debug
+#        command: /nsqd
         ports:
             - 4150:4150
             - 4151:4151
@@ -17,3 +18,7 @@ services:
         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

@@ -0,0 +1,43 @@
+<?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

@@ -0,0 +1,36 @@
+<?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

@@ -6,63 +6,29 @@ namespace Nsq;
 
 use PHPinnacle\Buffer\ByteBuffer;
 
-final class Buffer
+/**
+ * @psalm-suppress
+ */
+final class Buffer extends ByteBuffer
 {
-    private ByteBuffer $buffer;
+    public function readUInt32LE(): int
-
-    public function __construct(string $initial = '')
     {
-        $this->buffer = new ByteBuffer($initial);
+        /** @phpstan-ignore-next-line  */
-    }
+        return unpack('V', $this->consume(4))[1];
-
-    public function append(string $data): self
-    {
-        $this->buffer->append($data);
-
-        return $this;
-    }
-
-    public function consumeSize(): int
-    {
-        /** @see Bytes::BYTES_SIZE */
-        return $this->buffer->consumeUint32();
-    }
-
-    public function consumeType(): int
-    {
-        /** @see Bytes::BYTES_TYPE */
-        return $this->buffer->consumeUint32();
     }
 
     public function consumeTimestamp(): int
     {
-        /** @see Bytes::BYTES_TIMESTAMP */
+        return $this->consumeUint64();
-        return $this->buffer->consumeInt64();
     }
 
     public function consumeAttempts(): int
     {
-        /** @see Bytes::BYTES_ATTEMPTS */
+        return $this->consumeUint16();
-        return $this->buffer->consumeUint16();
     }
 
-    public function consumeId(): string
+    public function consumeMessageID(): string
     {
-        return $this->buffer->consume(Bytes::BYTES_ID);
+        return $this->consume(16);
-    }
-
-    public function size(): int
-    {
-        return $this->buffer->size();
-    }
-
-    public function bytes(): string
-    {
-        return $this->buffer->bytes();
-    }
-
-    public function flush(): string
-    {
-        return $this->buffer->flush();
     }
 }

src/Command.php

@@ -0,0 +1,111 @@
+<?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/Connection.php

@@ -4,213 +4,136 @@ declare(strict_types=1);
 
 namespace Nsq;
 
+use Amp\Promise;
 use Nsq\Config\ClientConfig;
 use Nsq\Config\ConnectionConfig;
 use Nsq\Exception\AuthenticationRequired;
-use Nsq\Exception\BadResponse;
-use Nsq\Exception\ConnectionFail;
-use Nsq\Exception\NotConnected;
-use Nsq\Exception\NsqError;
 use Nsq\Exception\NsqException;
-use Nsq\Protocol\Error;
+use Nsq\Frame\Response;
-use Nsq\Protocol\Frame;
+use Nsq\Stream\GzipStream;
-use Nsq\Protocol\Message;
+use Nsq\Stream\NullStream;
-use Nsq\Protocol\Response;
+use Nsq\Stream\SnappyStream;
-use Nsq\Socket\DeflateSocket;
+use Nsq\Stream\SocketStream;
-use Nsq\Socket\RawSocket;
-use Nsq\Socket\SnappySocket;
-use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
+use function Amp\call;
 
 /**
  * @internal
  */
 abstract class Connection
 {
-    use LoggerAwareTrait;
+    protected Stream $stream;
-
-    protected ClientConfig $clientConfig;
-
-    private NsqSocket $socket;
-
-    private ConnectionConfig $connectionConfig;
-
-    private bool $closed = false;
 
     public function __construct(
         private string $address,
-        ClientConfig $clientConfig = null,
+        private ClientConfig $clientConfig,
-        LoggerInterface $logger = null,
+        private LoggerInterface $logger,
     ) {
-        $this->logger = $logger ?? new NullLogger();
+        $this->stream = new NullStream();
-        $this->clientConfig = $clientConfig ?? new ClientConfig();
+    }
 
-        $socket = new RawSocket($this->address, $this->logger);
+    public function __destruct()
-        $socket->write('  V2');
+    {
-
+        $this->close();
-        $this->socket = new NsqSocket($socket);
-
-        $this->connectionConfig = ConnectionConfig::fromArray(
-            $this
-                ->command('IDENTIFY', data: $this->clientConfig->toString())
-                ->readResponse()
-                ->toArray()
-        );
-
-        if ($this->connectionConfig->snappy) {
-            $this->socket = new NsqSocket(
-                new SnappySocket(
-                    $socket,
-                    $this->logger,
-                ),
-            );
-
-            $this->checkIsOK();
-        }
-
-        if ($this->connectionConfig->deflate) {
-            $this->socket = new NsqSocket(
-                new DeflateSocket(
-                    $socket,
-                ),
-            );
-
-            $this->checkIsOK();
-        }
-
-        if ($this->connectionConfig->authRequired) {
-            if (null === $this->clientConfig->authSecret) {
-                throw new AuthenticationRequired();
-            }
-
-            $authResponse = $this
-                ->command('AUTH', data: $this->clientConfig->authSecret)
-                ->readResponse()
-                ->toArray()
-            ;
-
-            $this->logger->info('Authorization response: '.http_build_query($authResponse));
-        }
     }
 
     /**
-     * Cleanly close your connection (no more messages are sent).
+     * @return Promise<void>
      */
+    public function connect(): Promise
+    {
+        return call(function (): \Generator {
+            $buffer = new Buffer();
+
+            /** @var SocketStream $stream */
+            $stream = yield SocketStream::connect($this->address);
+
+            yield $stream->write(Command::magic());
+            yield $stream->write(Command::identify($this->clientConfig->toString()));
+
+            /** @var Response $response */
+            $response = yield $this->response($stream, $buffer);
+            $connectionConfig = ConnectionConfig::fromArray($response->toArray());
+
+            if ($connectionConfig->snappy) {
+                $stream = new SnappyStream($stream, $buffer->flush());
+
+                /** @var Response $response */
+                $response = yield $this->response($stream, $buffer);
+
+                if (!$response->isOk()) {
+                    throw new NsqException();
+                }
+            }
+
+            if ($connectionConfig->deflate) {
+                $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();
+                }
+
+                yield $stream->write(Command::auth($this->clientConfig->authSecret));
+
+                /** @var Response $response */
+                $response = yield $this->response($stream, $buffer);
+
+                $this->logger->info('Authorization response: '.http_build_query($response->toArray()));
+            }
+
+            $this->stream = $stream;
+        });
+    }
+
     public function close(): void
     {
-        if ($this->closed) {
+//        $this->stream->write(Command::cls());
-            return;
-        }
 
-        try {
+        $this->stream->close();
-            $this->command('CLS');
+        $this->stream = new NullStream();
-            $this->socket->close();
-        } catch (\Throwable $e) {
-        }
-
-        $this->closed = true;
     }
 
-    public function isClosed(): bool
+    protected function handleError(Frame\Error $error): void
     {
-        return $this->closed;
+        $this->logger->error($error->data);
+
+        if (ErrorType::terminable($error)) {
+            $this->close();
+
+            throw $error->toException();
+        }
     }
 
     /**
-     * @param array<int, int|string>|string $params
+     * @return Promise<Frame\Response>
      */
-    protected function command(string $command, array | string $params = [], string $data = null): self
+    private function response(Stream $stream, Buffer $buffer): Promise
     {
-        if ($this->closed) {
+        return call(function () use ($stream, $buffer): \Generator {
-            throw new NotConnected('Connection closed.');
+            while (true) {
-        }
+                $response = Parser::parse($buffer);
 
-        $command = [] === $params
+                if (null === $response && null !== ($chunk = yield $stream->read())) {
-            ? $command
+                    $buffer->append($chunk);
-            : implode(' ', [$command, ...((array) $params)]);
 
-        $this->logger->info('Command [{command}] with data [{data}]', ['command' => $command, 'data' => $data ?? 'null']);
+                    continue;
+                }
 
-        $this->socket->write($command, $data);
+                if (!$response instanceof Frame\Response) {
+                    throw new NsqException();
+                }
 
-        return $this;
+                return $response;
-    }
-
-    public function hasMessage(float $timeout): bool
-    {
-        if ($this->closed) {
-            throw new NotConnected('Connection closed.');
-        }
-
-        try {
-            return false !== $this->socket->wait($timeout);
-        } catch (ConnectionFail $e) {
-            $this->close();
-
-            throw $e;
-        }
-    }
-
-    protected function readFrame(): Frame
-    {
-        if ($this->closed) {
-            throw new NotConnected('Connection closed.');
-        }
-
-        $buffer = $this->socket->read();
-
-        $this->logger->debug('Received buffer: '.addcslashes($buffer->bytes(), PHP_EOL));
-
-        return match ($type = $buffer->consumeType()) {
-            0 => new Response($buffer->flush()),
-            1 => new Error($buffer->flush()),
-            2 => new Message(
-                timestamp: $buffer->consumeTimestamp(),
-                attempts: $buffer->consumeAttempts(),
-                id: $buffer->consumeId(),
-                body: $buffer->flush(),
-                consumer: $this instanceof Consumer ? $this : throw new NsqException('what?'),
-            ),
-            default => throw new NsqException('Unexpected frame type: '.$type)
-        };
-    }
-
-    protected function checkIsOK(): void
-    {
-        $response = $this->readResponse();
-
-        if ($response->isHeartBeat()) {
-            $this->command('NOP');
-
-            $this->checkIsOK();
-
-            return;
-        }
-
-        if (!$response->isOk()) {
-            throw new BadResponse($response);
-        }
-
-        $this->logger->info('Ok checked.');
-    }
-
-    private function readResponse(): Response
-    {
-        $frame = $this->readFrame();
-
-        if ($frame instanceof Response) {
-            return $frame;
-        }
-
-        if ($frame instanceof Error) {
-            if ($frame->type->terminateConnection) {
-                $this->close();
             }
-
+        });
-            throw new NsqError($frame);
-        }
-
-        throw new NsqException('Unreachable statement.');
     }
 }

src/Consumer.php

@@ -4,108 +4,156 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Generator;
+use Amp\Failure;
+use Amp\Promise;
 use Nsq\Config\ClientConfig;
-use Nsq\Exception\NsqError;
+use Nsq\Exception\ConsumerException;
-use Nsq\Exception\NsqException;
+use Nsq\Frame\Response;
-use Nsq\Protocol\Error;
+use Nsq\Stream\NullStream;
-use Nsq\Protocol\Message;
-use Nsq\Protocol\Response;
 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
+     */
+    private $onMessage;
+
     public function __construct(
+        private string $address,
         private string $topic,
         private string $channel,
-        string $address,
+        callable $onMessage,
-        ClientConfig $clientConfig = null,
+        ClientConfig $clientConfig,
-        LoggerInterface $logger = null
+        private LoggerInterface $logger,
     ) {
-        parent::__construct($address, $clientConfig, $logger);
+        parent::__construct(
+            $this->address,
+            $clientConfig,
+            $this->logger,
+        );
+
+        $this->onMessage = $onMessage;
     }
 
-    /**
+    public static function create(
-     * @psalm-return Generator<int, Message|float|null, int|null, void>
+        string $address,
-     */
+        string $topic,
-    public function generator(): \Generator
+        string $channel,
-    {
+        callable $onMessage,
-        $this->command('SUB', [$this->topic, $this->channel])->checkIsOK();
+        ?ClientConfig $clientConfig = null,
-
+        ?LoggerInterface $logger = null,
-        while (true) {
+    ): self {
-            $this->rdy(1);
+        return new self(
-
+            $address,
-            $timeout = $this->clientConfig->readTimeout;
+            $topic,
-
+            $channel,
-            do {
+            $onMessage,
-                $deadline = microtime(true) + $timeout;
+            $clientConfig ?? new ClientConfig(),
-
+            $logger ?? new NullLogger(),
-                $message = $this->hasMessage($timeout) ? $this->readMessage() : null;
+        );
-
-                $timeout = ($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime;
-            } while (0 < $timeout && null === $message);
-
-            $command = yield $message;
-
-            if (0 === $command) {
-                break;
-            }
-        }
-
-        $this->close();
     }
 
-    public function readMessage(): ?Message
+    public function connect(): Promise
     {
-        $frame = $this->readFrame();
+        if (!$this->stream instanceof NullStream) {
-
+            return call(static function (): void {
-        if ($frame instanceof Message) {
+            });
-            return $frame;
         }
 
-        if ($frame instanceof Response && $frame->isHeartBeat()) {
+        return call(function (): \Generator {
-            $this->command('NOP');
+            yield parent::connect();
 
-            return null;
+            $this->run();
-        }
+        });
+    }
 
-        if ($frame instanceof Error) {
+    private function run(): void
-            if ($frame->type->terminateConnection) {
+    {
-                $this->close();
+        $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);
             }
 
-            throw new NsqError($frame);
+            /** @var Response $response */
-        }
+            $response = Parser::parse($buffer);
 
-        throw new NsqException('Unreachable statement.');
+            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): void
+    public function rdy(int $count): Promise
     {
         if ($this->rdy === $count) {
-            return;
+            return call(static function (): void {
+            });
         }
 
-        $this->command('RDY', (string) $count);
-
         $this->rdy = $count;
+
+        return $this->stream->write(Command::rdy($count));
     }
 
     /**
      * Finish a message (indicate successful processing).
      *
+     * @return Promise<void>
+     *
      * @internal
      */
-    public function fin(string $id): void
+    public function fin(string $id): Promise
     {
-        $this->command('FIN', $id);
-
         --$this->rdy;
+
+        return $this->stream->write(Command::fin($id));
     }
 
     /**
@@ -114,22 +162,26 @@ final class Consumer extends Connection
      * be explicitly relied upon and may change in the future. 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): void
+    public function req(string $id, int $timeout): Promise
     {
-        $this->command('REQ', [$id, $timeout]);
-
         --$this->rdy;
+
+        return $this->stream->write(Command::req($id, $timeout));
     }
 
     /**
      * Reset the timeout for an in-flight message.
      *
+     * @return Promise<void>
+     *
      * @internal
      */
-    public function touch(string $id): void
+    public function touch(string $id): Promise
     {
-        $this->command('TOUCH', $id);
+        return $this->stream->write(Command::touch($id));
     }
 }

src/ErrorType.php

@@ -2,7 +2,7 @@
 
 declare(strict_types=1);
 
-namespace Nsq\Protocol;
+namespace Nsq;
 
 /**
  * @psalm-immutable
@@ -88,13 +88,12 @@ final class ErrorType
      */
     public const E_UNAUTHORIZED = true;
 
-    /**
+    public static function terminable(Frame\Error $error): bool
-     * A boolean indicating whether or not an [Error] with this type terminates the connection or not.
-     */
-    public bool $terminateConnection;
-
-    public function __construct(public string $type)
     {
-        $this->terminateConnection = \constant('self::'.$this->type) ?? self::E_INVALID;
+        $type = explode(' ', $error->data)[0];
+
+        $constant = 'self::'.$type;
+
+        return \defined($constant) ? \constant($constant) : self::E_INVALID;
     }
 }

src/Exception/BadResponse.php

@@ -1,15 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-use Nsq\Protocol\Response;
-
-final class BadResponse extends NsqException
-{
-    public function __construct(Response $response)
-    {
-        parent::__construct($response->msg);
-    }
-}

src/Exception/ConnectionFail.php

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

src/Exception/ConsumerException.php

@@ -0,0 +1,15 @@
+<?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

@@ -1,25 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-use Nsq\Protocol\Message;
-
-final class MessageAlreadyFinished extends 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

@@ -0,0 +1,15 @@
+<?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/NotConnected.php

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

src/Exception/NsqError.php

@@ -1,15 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-use Nsq\Protocol\Error;
-
-final class NsqError extends NsqException
-{
-    public function __construct(Error $error)
-    {
-        parent::__construct($error->rawData);
-    }
-}

src/Exception/ServerException.php

@@ -4,6 +4,6 @@ declare(strict_types=1);
 
 namespace Nsq\Exception;
 
-final class NullReceived extends NsqException
+final class ServerException extends NsqException
 {
 }

src/Exception/SnappyException.php

@@ -0,0 +1,18 @@
+<?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/Frame.php

@@ -0,0 +1,32 @@
+<?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

@@ -0,0 +1,24 @@
+<?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

@@ -0,0 +1,19 @@
+<?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

@@ -2,9 +2,9 @@
 
 declare(strict_types=1);
 
-namespace Nsq\Protocol;
+namespace Nsq\Frame;
 
-use Nsq\Bytes;
+use Nsq\Frame;
 
 /**
  * @psalm-immutable
@@ -14,19 +14,19 @@ final class Response extends Frame
     public const OK = 'OK';
     public const HEARTBEAT = '_heartbeat_';
 
-    public function __construct(public string $msg)
+    public function __construct(public string $data)
     {
-        parent::__construct(\strlen($this->msg) + Bytes::BYTES_TYPE);
+        parent::__construct(self::TYPE_RESPONSE);
     }
 
     public function isOk(): bool
     {
-        return self::OK === $this->msg;
+        return self::OK === $this->data;
     }
 
     public function isHeartBeat(): bool
     {
-        return self::HEARTBEAT === $this->msg;
+        return self::HEARTBEAT === $this->data;
     }
 
     /**
@@ -34,6 +34,6 @@ final class Response extends Frame
      */
     public function toArray(): array
     {
-        return json_decode($this->msg, true, flags: JSON_THROW_ON_ERROR);
+        return json_decode($this->data, true, flags: JSON_THROW_ON_ERROR);
     }
 }

src/Message.php

@@ -0,0 +1,82 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq;
+
+use Amp\Promise;
+use Nsq\Exception\MessageException;
+use function Amp\call;
+
+final class Message
+{
+    private bool $processed = false;
+
+    public function __construct(
+        public string $id,
+        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>
+     */
+    public function finish(): Promise
+    {
+        return call(function (): \Generator {
+            if ($this->processed) {
+                throw MessageException::processed($this);
+            }
+
+            yield $this->consumer->fin($this->id);
+
+            $this->processed = true;
+        });
+    }
+
+    /**
+     * @return Promise<void>
+     */
+    public function requeue(int $timeout): Promise
+    {
+        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>
+     */
+    public function touch(): Promise
+    {
+        return call(function (): \Generator {
+            if ($this->processed) {
+                throw MessageException::processed($this);
+            }
+
+            yield $this->consumer->touch($this->id);
+
+            $this->processed = true;
+        });
+    }
+}

src/NsqSocket.php

@@ -1,78 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq;
-
-use Nsq\Exception\ConnectionFail;
-use Nsq\Socket\Socket;
-use PHPinnacle\Buffer\ByteBuffer;
-use Throwable;
-
-final class NsqSocket
-{
-    private Buffer $input;
-
-    private ByteBuffer $output;
-
-    public function __construct(
-        private Socket $socket,
-    ) {
-        $this->input = new Buffer();
-        $this->output = new ByteBuffer();
-    }
-
-    public function write(string $command, string $data = null): void
-    {
-        $this->output->append($command.PHP_EOL);
-
-        if (null !== $data) {
-            $this->output->appendUint32(\strlen($data));
-            $this->output->append($data);
-        }
-
-        $this->socket->write($this->output->flush());
-    }
-
-    public function wait(float $timeout): bool
-    {
-        return $this->socket->selectRead($timeout);
-    }
-
-    public function read(): Buffer
-    {
-        $buffer = $this->input;
-
-        $size = Bytes::BYTES_SIZE;
-
-        do {
-            $buffer->append(
-                $this->socket->read($size),
-            );
-
-            $size -= $buffer->size();
-        } while ($buffer->size() < Bytes::BYTES_SIZE);
-
-        if ('' === $buffer->bytes()) {
-            throw new ConnectionFail('Probably connection closed.');
-        }
-
-        $size = $buffer->consumeSize();
-
-        do {
-            $buffer->append(
-                $this->socket->read($size - $buffer->size()),
-            );
-        } while ($buffer->size() < $size);
-
-        return $buffer;
-    }
-
-    public function close(): void
-    {
-        try {
-            $this->socket->close();
-        } catch (Throwable) {
-        }
-    }
-}

src/Parser.php

@@ -0,0 +1,47 @@
+<?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,38 +4,96 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use PHPinnacle\Buffer\ByteBuffer;
+use Amp\Promise;
+use Nsq\Config\ClientConfig;
+use Nsq\Exception\NsqException;
+use Nsq\Stream\NullStream;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use function Amp\asyncCall;
+use function Amp\call;
 
-/**
- * @psalm-suppress PropertyNotSetInConstructor
- */
 final class Producer extends Connection
 {
-    public function pub(string $topic, string $body): void
+    public static function create(
+        string $address,
+        ClientConfig $clientConfig = null,
+        LoggerInterface $logger = null,
+    ): self {
+        return new self(
+            $address,
+            $clientConfig ?? new ClientConfig(),
+            $logger ?? new NullLogger(),
+        );
+    }
+
+    public function connect(): Promise
     {
-        $this->command('PUB', $topic, $body)->checkIsOK();
+        if (!$this->stream instanceof NullStream) {
+            return call(static function (): void {
+            });
+        }
+
+        return call(function (): \Generator {
+            yield parent::connect();
+
+            $this->run();
+        });
     }
 
     /**
-     * @psalm-param array<int, mixed> $bodies
+     * @param array<int, string>|string $body
+     *
+     * @return Promise<void>
      */
-    public function mpub(string $topic, array $bodies): void
+    public function publish(string $topic, string | array $body, int $delay = 0): Promise
     {
-        static $buffer;
+        if (0 < $delay) {
-        $buffer ??= new ByteBuffer();
+            return call(
-
+                function (array $bodies) use ($topic, $delay): \Generator {
-        $buffer->appendUint32(\count($bodies));
+                    foreach ($bodies as $body) {
-
+                        yield $this->stream->write(Command::dpub($topic, $body, $delay));
-        foreach ($bodies as $body) {
+                    }
-            $buffer->appendUint32(\strlen($body));
+                },
-            $buffer->append($body);
+                (array) $body,
+            );
         }
 
-        $this->command('MPUB', $topic, $buffer->flush())->checkIsOK();
+        $command = \is_array($body)
+            ? Command::mpub($topic, $body)
+            : Command::pub($topic, $body);
+
+        return $this->stream->write($command);
     }
 
-    public function dpub(string $topic, string $body, int $delay): void
+    private function run(): void
     {
-        $this->command('DPUB', [$topic, $delay], $body)->checkIsOK();
+        $buffer = new Buffer();
+
+        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/Protocol/Error.php

@@ -1,22 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Protocol;
-
-use Nsq\Bytes;
-
-/**
- * @psalm-immutable
- */
-final class Error extends Frame
-{
-    public ErrorType $type;
-
-    public function __construct(public string $rawData)
-    {
-        parent::__construct(\strlen($this->rawData) + Bytes::BYTES_TYPE);
-
-        $this->type = new ErrorType(explode(' ', $this->rawData)[0]);
-    }
-}

src/Protocol/Frame.php

@@ -1,16 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Protocol;
-
-abstract class Frame
-{
-    public function __construct(
-        /**
-         * @psalm-readonly
-         */
-        public int $length,
-    ) {
-    }
-}

src/Protocol/Message.php

@@ -1,88 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Protocol;
-
-use Nsq\Bytes;
-use Nsq\Consumer;
-use Nsq\Exception\MessageAlreadyFinished;
-
-final class Message extends Frame
-{
-    /**
-     * @psalm-readonly
-     */
-    public int $timestamp;
-
-    /**
-     * @psalm-readonly
-     */
-    public int $attempts;
-
-    /**
-     * @psalm-readonly
-     */
-    public string $id;
-
-    /**
-     * @psalm-readonly
-     */
-    public string $body;
-
-    private bool $finished = false;
-
-    private Consumer $consumer;
-
-    public function __construct(int $timestamp, int $attempts, string $id, string $body, Consumer $consumer)
-    {
-        parent::__construct(
-            Bytes::BYTES_TYPE
-            + Bytes::BYTES_TIMESTAMP
-            + Bytes::BYTES_ATTEMPTS
-            + Bytes::BYTES_ID
-            + \strlen($body)
-        );
-
-        $this->timestamp = $timestamp;
-        $this->attempts = $attempts;
-        $this->id = $id;
-        $this->body = $body;
-
-        $this->consumer = $consumer;
-    }
-
-    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/Reconnect/ExponentialStrategy.php

@@ -1,62 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Reconnect;
-
-use Nsq\Exception\ConnectionFail;
-use Psr\Log\LoggerAwareTrait;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
-
-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

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

src/Reconnect/ReconnectStrategy.php

@@ -1,15 +0,0 @@
-<?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

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

src/Socket/DeflateSocket.php

@@ -1,45 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Socket;
-
-final class DeflateSocket implements Socket
-{
-    public function __construct(
-        private Socket $socket,
-    ) {
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function write(string $data): void
-    {
-        throw new \LogicException('not implemented.');
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function read(int $length): string
-    {
-        throw new \LogicException('not implemented.');
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function close(): void
-    {
-        throw new \LogicException('not implemented.');
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function selectRead(float $timeout): bool
-    {
-        return $this->socket->selectRead($timeout);
-    }
-}

src/Socket/RawSocket.php

@@ -1,81 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Socket;
-
-use Nsq\Exception\ConnectionFail;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
-use Socket\Raw\Exception;
-use Socket\Raw\Factory;
-use Socket\Raw\Socket as ClueSocket;
-use Throwable;
-
-final class RawSocket implements Socket
-{
-    private ClueSocket $socket;
-
-    private LoggerInterface $logger;
-
-    public function __construct(string $address, LoggerInterface $logger = null)
-    {
-        $this->socket = (new Factory())->createClient($address);
-        $this->logger = $logger ?? new NullLogger();
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function selectRead(float $timeout): bool
-    {
-        try {
-            return false !== $this->socket->selectRead($timeout);
-        } // @codeCoverageIgnoreStart
-        catch (Exception $e) {
-            throw ConnectionFail::fromThrowable($e);
-        }
-        // @codeCoverageIgnoreEnd
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function close(): void
-    {
-        try {
-            $this->socket->close();
-        } catch (Throwable) {
-        }
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function write(string $data): void
-    {
-        try {
-            $this->socket->write($data);
-        } // @codeCoverageIgnoreStart
-        catch (Exception $e) {
-            $this->logger->error($e->getMessage(), ['exception' => $e]);
-
-            throw ConnectionFail::fromThrowable($e);
-        }
-        // @codeCoverageIgnoreEnd
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function read(int $length): string
-    {
-        try {
-            return $this->socket->read($length);
-        } // @codeCoverageIgnoreStart
-        catch (Exception $e) {
-            throw ConnectionFail::fromThrowable($e);
-        }
-        // @codeCoverageIgnoreEnd
-    }
-}

src/Socket/SnappySocket.php

@@ -1,162 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Socket;
-
-use PHPinnacle\Buffer\ByteBuffer;
-use Psr\Log\LoggerInterface;
-
-final class SnappySocket implements Socket
-{
-    private ByteBuffer $output;
-
-    private ByteBuffer $input;
-
-    public function __construct(
-        private Socket $socket,
-        private LoggerInterface $logger,
-    ) {
-        if (
-            !\function_exists('snappy_compress')
-            || !\function_exists('snappy_uncompress')
-        || !\extension_loaded('snappy')
-        ) {
-            throw new \LogicException('Snappy extension not installed.');
-        }
-
-        $this->output = new ByteBuffer();
-        $this->input = new ByteBuffer();
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function write(string $data): void
-    {
-        $identifierFrame = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
-        $compressedFrame = 0x00;
-        $uncompressedFrame = 0x01; // 11
-        $maxChunkLength = 65536;
-
-        $byteBuffer = new ByteBuffer();
-        foreach ($identifierFrame as $bite) {
-            $byteBuffer->appendUint8($bite);
-        }
-
-        foreach (str_split($data, $maxChunkLength) as $chunk) {
-            $compressedChunk = snappy_compress($chunk);
-
-            [$chunk, $chunkType] = \strlen($compressedChunk) <= 0.875 * \strlen($data)
-                ? [$compressedChunk, $compressedFrame]
-                : [$data, $uncompressedFrame];
-
-            /** @var string $checksum */
-            $checksum = hash('crc32c', $data, true);
-            /** @phpstan-ignore-next-line */
-            $checksum = unpack('N', $checksum)[1];
-            $maskedChecksum = (($checksum >> 15) | ($checksum << 17)) + 0xa282ead8 & 0xffffffff;
-
-            $size = (\strlen($chunk) + 4) << 8;
-
-            $byteBuffer->append(pack('V', $chunkType + $size));
-            $byteBuffer->append(pack('V', $maskedChecksum));
-            $byteBuffer->append($chunk);
-        }
-
-        $this->socket->write($byteBuffer->flush());
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function read(int $length): string
-    {
-        $output = $this->output;
-        $input = $this->input;
-
-        $this->logger->debug('Snappy requested {length} bytes.', ['length' => $length]);
-
-        while ($output->size() < $length) {
-            $this->logger->debug('Snappy enter loop');
-
-            /** @phpstan-ignore-next-line */
-            $chunkType = unpack('V', $this->socket->read(4))[1];
-
-            $size = $chunkType >> 8;
-            $chunkType &= 0xff;
-
-            $this->logger->debug('Snappy receive chunk [{chunk}], size [{size}]', [
-                'chunk' => $chunkType,
-                'size' => $size,
-            ]);
-
-            do {
-                $input->append(
-                    $this->socket->read($size),
-                );
-
-                $size -= $input->size();
-            } while ($input->size() < $size);
-
-            switch ($chunkType) {
-                case 0xff:
-                    $this->logger->debug('Snappy identifier chunk');
-
-                    $input->discard(6); // discard identifier body
-
-                    break;
-                case 0x00: // 'compressed',
-                    $this->logger->debug('Snappy compressed chunk');
-
-                    $data = $input
-                        ->discard(4) // discard checksum
-                        ->flush()
-                    ;
-
-                    $this->logger->debug('Snappy compressed data [{data}]', ['data' => $data]);
-
-                    $output->append(snappy_uncompress($data));
-
-                    break;
-                case 0x01: // 'uncompressed',
-                    $this->logger->debug('Snappy uncompressed chunk');
-
-                    $data = $input
-                        ->discard(4) // discard checksum
-                        ->flush()
-                    ;
-
-                    $this->logger->debug('Snappy uncompressed data [{data}]', ['data' => $data]);
-
-                    $output->append($data);
-
-                    break;
-                case 0xfe:// 'padding',
-                    $this->logger->debug('Snappy padding chunk');
-
-                    break;
-            }
-        }
-
-        $this->logger->debug('Snappy return message [{message}]', ['message' => $output->read($length)]);
-
-        return $output->consume($length);
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function close(): void
-    {
-        $this->socket->close();
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function selectRead(float $timeout): bool
-    {
-        return !$this->input->empty() || $this->socket->selectRead($timeout);
-    }
-}

src/Socket/Socket.php

@@ -1,27 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Socket;
-
-use Nsq\Exception\ConnectionFail;
-
-interface Socket
-{
-    /**
-     * @throws ConnectionFail
-     */
-    public function write(string $data): void;
-
-    /**
-     * @throws ConnectionFail
-     */
-    public function read(int $length): string;
-
-    /**
-     * @throws ConnectionFail
-     */
-    public function selectRead(float $timeout): bool;
-
-    public function close(): void;
-}

src/Stream.php

@@ -0,0 +1,22 @@
+<?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

@@ -0,0 +1,38 @@
+<?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

@@ -0,0 +1,37 @@
+<?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

@@ -0,0 +1,115 @@
+<?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

@@ -0,0 +1,64 @@
+<?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();
+    }
+}

tests/ErrorTypeTest.php

@@ -0,0 +1,28 @@
+<?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

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

@@ -3,10 +3,6 @@
 declare(strict_types=1);
 
 use Nsq\Config\ClientConfig;
-use Nsq\Consumer;
-use Nsq\Producer;
-use Nsq\Protocol\Message;
-use Nyholm\NSA;
 use PHPUnit\Framework\TestCase;
 
 final class NsqTest extends TestCase
@@ -16,78 +12,7 @@ final class NsqTest extends TestCase
      */
     public function test(ClientConfig $clientConfig): void
     {
-        $producer = new Producer('tcp://localhost:4150');
+        self::markTestSkipped('');
-        $producer->pub(__FUNCTION__, __FUNCTION__);
-
-        $consumer = new Consumer(
-            topic: 'test',
-            channel: 'test',
-            address: 'tcp://localhost:4150',
-            clientConfig: $clientConfig,
-        );
-        $generator = $consumer->generator();
-
-        /** @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::assertFalse($consumer->isClosed());
-        $generator->send(0);
-        self::assertTrue($consumer->isClosed());
     }
 
     /**

tests/ProducerTest.php

@@ -2,7 +2,8 @@
 
 declare(strict_types=1);
 
-use Nsq\Exception\NsqError;
+use Amp\Loop;
+use Nsq\Exception\ServerException;
 use Nsq\Producer;
 use PHPUnit\Framework\TestCase;
 
@@ -13,11 +14,16 @@ final class ProducerTest extends TestCase
      */
     public function testPubFail(string $topic, string $body, string $exceptionMessage): void
     {
-        $this->expectException(NsqError::class);
+        $this->expectException(ServerException::class);
         $this->expectExceptionMessage($exceptionMessage);
 
-        $producer = new Producer('tcp://localhost:4150');
+        $producer = Producer::create('tcp://localhost:4150');
-        $producer->pub($topic, $body);
+
+        Loop::run(static function () use ($producer, $topic, $body): Generator {
+            yield $producer->connect();
+
+            yield $producer->publish($topic, $body);
+        });
     }
 
     /**

tests/Protocol/ErrorTypeTest.php

@@ -1,30 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Protocol;
-
-use Nsq\Protocol\ErrorType;
-use PHPUnit\Framework\TestCase;
-
-final class ErrorTypeTest extends TestCase
-{
-    /**
-     * @dataProvider data
-     */
-    public function testConstructor(string $type, bool $isConnectionTerminated): void
-    {
-        $errorType = new ErrorType($type);
-        self::assertSame($isConnectionTerminated, $errorType->terminateConnection);
-    }
-
-    /**
-     * @return \Generator<string, array<int, bool|string>>
-     */
-    public function data(): \Generator
-    {
-        foreach ((new \ReflectionClass(ErrorType::class))->getConstants() as $constant => $isTerminated) {
-            yield $constant => [$constant, $isTerminated];
-        }
-    }
-}