rename test data

.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
 
@@ -58,6 +59,9 @@ 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
 
@@ -108,6 +112,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 +144,7 @@ jobs:
                 uses: shivammathur/setup-php@v2
                 with:
                     php-version: '8.0'
+                    extensions: snappy-kjdev/php-ext-snappy@0.2.1
                 env:
                     update: true
 
@@ -194,6 +200,9 @@ 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,3 +4,14 @@
 /.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

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

@@ -43,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"
@@ -60,7 +60,7 @@
             "vendor/bin/psalm"
         ],
         "test": [
-            "@norm-check",
+            "@norm",
             "@cs",
             "@phpstan",
             "@psalm",

docker-compose.yml

@@ -18,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

@@ -9,12 +9,11 @@ use Amp\Log\ConsoleFormatter;
 use Amp\Log\StreamHandler;
 use Amp\Loop;
 use Amp\Promise;
-use Amp\Success;
 use Monolog\Logger;
 use Monolog\Processor\PsrLogMessageProcessor;
 use Nsq\Config\ClientConfig;
 use Nsq\Consumer;
-use Nsq\Protocol\Message;
+use Nsq\Message;
 use function Amp\call;
 
 Loop::run(static function () {
@@ -24,16 +23,6 @@ Loop::run(static function () {
 
     $consumer = new Consumer(
         'tcp://localhost:4150',
-        clientConfig: new ClientConfig(
-            deflate: false,
-            snappy: false,
-        ),
-        logger: $logger,
-    );
-
-    yield $consumer->connect();
-
-    yield $consumer->listen(
         topic: 'local',
         channel: 'local',
         onMessage: static function (Message $message) use ($logger): Promise {
@@ -41,9 +30,14 @@ Loop::run(static function () {
                 $logger->info('Received: {body}', ['body' => $message->body]);
 
                 yield $message->finish();
-
-                return new Success(false);
             });
-        }
+        },
+        clientConfig: new ClientConfig(
+            deflate: false,
+            snappy: true,
+        ),
+        logger: $logger,
     );
+
+    yield $consumer->connect();
 });

examples/producer.php

@@ -22,12 +22,15 @@ Loop::run(static function () {
         'tcp://localhost:4150',
         clientConfig: new ClientConfig(
             deflate: false,
-            snappy: false,
+            heartbeatInterval: 5000,
+            snappy: true,
         ),
         logger: $logger,
     );
 
     yield $producer->connect();
 
-    yield $producer->pub(topic: 'topic', body: 'Message body!');
+    while (true) {
+        yield $producer->publish(topic: 'local', body: array_fill(0, 200, 'Message body!'));
+    }
 });

src/Buffer.php

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

@@ -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,63 +4,32 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use Amp\ByteStream\InputStream;
-use Amp\ByteStream\OutputStream;
-use Amp\ByteStream\ZlibInputStream;
-use Amp\ByteStream\ZlibOutputStream;
-use Amp\Failure;
 use Amp\Promise;
-use Amp\Socket\Socket;
-use Amp\Success;
 use Nsq\Config\ClientConfig;
 use Nsq\Config\ConnectionConfig;
 use Nsq\Exception\AuthenticationRequired;
-use Nsq\Exception\BadResponse;
-use Nsq\Exception\NotConnected;
-use Nsq\Exception\NsqError;
 use Nsq\Exception\NsqException;
-use Nsq\Protocol\Error;
-use Nsq\Protocol\Frame;
-use Nsq\Protocol\Message;
-use Nsq\Protocol\Response;
-use Nsq\Stream\NsqInputStream;
+use Nsq\Frame\Response;
+use Nsq\Stream\GzipStream;
 use Nsq\Stream\NullStream;
-use Nsq\Stream\SnappyInputStream;
-use Nsq\Stream\SnappyOutputStream;
-use PHPinnacle\Buffer\ByteBuffer;
+use Nsq\Stream\SnappyStream;
+use Nsq\Stream\SocketStream;
 use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
 use function Amp\call;
-use function Amp\Socket\connect;
 
 /**
  * @internal
  */
 abstract class Connection
 {
-    private ?Socket $socket = null;
+    protected Stream $stream;
 
-    private InputStream $inputStream;
-
-    private OutputStream $outputStream;
-
-    private ByteBuffer $buffer;
-
-    protected ?ConnectionConfig $connectionConfig = null;
-
-    protected ClientConfig $clientConfig;
-
-    protected LoggerInterface $logger;
-
-    final public function __construct(
+    public function __construct(
         private string $address,
-        ClientConfig $clientConfig = null,
-        ?LoggerInterface $logger = null,
+        private ClientConfig $clientConfig,
+        private LoggerInterface $logger,
     ) {
-        $this->buffer = new ByteBuffer();
-        $this->inputStream = $this->outputStream = new NullStream();
-        $this->clientConfig = $clientConfig ?? new ClientConfig();
-        $this->logger = $logger ?? new NullLogger();
+        $this->stream = new NullStream();
     }
 
     public function __destruct()
@@ -74,184 +43,97 @@ abstract class Connection
     public function connect(): Promise
     {
         return call(function (): \Generator {
-            $this->socket = $this->outputStream = yield connect($this->address);
-            $this->inputStream = new NsqInputStream($this->socket);
+            $buffer = new Buffer();
 
-            yield $this->outputStream->write('  V2');
+            /** @var SocketStream $stream */
+            $stream = yield SocketStream::connect($this->address);
+
+            yield $stream->write(Command::magic());
+            yield $stream->write(Command::identify($this->clientConfig->toString()));
 
-            yield $this->command('IDENTIFY', data: $this->clientConfig->toString());
             /** @var Response $response */
-            $response = yield $this->readResponse();
-            $this->connectionConfig = ConnectionConfig::fromArray($response->toArray());
+            $response = yield $this->response($stream, $buffer);
+            $connectionConfig = ConnectionConfig::fromArray($response->toArray());
 
-            if ($this->connectionConfig->snappy) {
-                $this->inputStream = new NsqInputStream(
-                    new SnappyInputStream($this->inputStream, $this->logger),
-                );
-                $this->outputStream = new SnappyOutputStream($this->outputStream);
+            if ($connectionConfig->snappy) {
+                $stream = new SnappyStream($stream, $buffer->flush());
 
-                $this->checkIsOK();
+                /** @var Response $response */
+                $response = yield $this->response($stream, $buffer);
+
+                if (!$response->isOk()) {
+                    throw new NsqException();
+                }
             }
 
-            if ($this->connectionConfig->deflate) {
-                $this->inputStream = new NsqInputStream(
-                    new ZlibInputStream($this->socket, ZLIB_ENCODING_DEFLATE, [
-                        'level' => $this->connectionConfig->deflateLevel,
-                    ]),
-                );
-                $this->outputStream = new ZlibOutputStream($this->socket, ZLIB_ENCODING_DEFLATE, [
-                    'level' => $this->connectionConfig->deflateLevel,
-                ]);
+            if ($connectionConfig->deflate) {
+                $stream = new GzipStream($stream);
 
-                $this->checkIsOK();
+                /** @var Response $response */
+                $response = yield $this->response($stream, $buffer);
+
+                if (!$response->isOk()) {
+                    throw new NsqException();
+                }
             }
 
-            if ($this->connectionConfig->authRequired) {
+            if ($connectionConfig->authRequired) {
                 if (null === $this->clientConfig->authSecret) {
-                    yield $this->close();
-
                     throw new AuthenticationRequired();
                 }
 
-                yield $this->command('AUTH', data: $this->clientConfig->authSecret);
-                $response = yield $this->readResponse();
+                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;
         });
     }
 
-    /**
-     * Cleanly close your connection (no more messages are sent).
-     *
-     * @return Promise<void>
-     */
-    public function close(): Promise
+    public function close(): void
     {
-        if (null === $this->socket) {
-            return new Success();
+//        $this->stream->write(Command::cls());
+
+        $this->stream->close();
+        $this->stream = new NullStream();
     }
 
-        return call(function (): \Generator {
-            yield $this->command('CLS');
-
-            if (null !== $this->socket) {
-                $this->socket->close();
-
-                $this->socket = null;
-            }
-        });
-    }
-
-    public function isClosed(): bool
+    protected function handleError(Frame\Error $error): void
     {
-        return null === $this->socket;
-    }
+        $this->logger->error($error->data);
 
-    /**
-     * @param array<int, int|string>|string $params
-     *
-     * @return Promise<void>
-     */
-    protected function command(string $command, array | string $params = [], string $data = null): Promise
-    {
-        if (null === $this->socket) {
-            return new Failure(new NotConnected());
-        }
-
-        $command = implode(' ', [$command, ...((array) $params)]);
-
-        $buffer = $this->buffer->append($command.PHP_EOL);
-
-        if (null !== $data) {
-            $buffer->appendUint32(\strlen($data));
-            $buffer->append($data);
-        }
-
-        $this->logger->debug('Sending: {bytes}', ['bytes' => $buffer->bytes()]);
-
-        return $this->outputStream->write($buffer->flush());
-    }
-
-    /**
-     * @return Promise<Frame>
-     */
-    protected function readFrame(): Promise
-    {
-        return call(function (): \Generator {
-            $bytes = yield $this->inputStream->read();
-
-            $this->logger->debug('Receiving: {bytes}', ['bytes' => $bytes]);
-
-            if (null === $bytes) {
-                throw new NotConnected();
-            }
-
-            $buffer = $this->buffer->append($bytes);
-
-            $frame = match ($type = $buffer->consumeUint32()) {
-                0 => new Response($buffer->flush()),
-                1 => new Error($buffer->flush()),
-                2 => new Message(
-                    timestamp: $buffer->consumeInt64(),
-                    attempts: $buffer->consumeUint16(),
-                    id: $buffer->consume(Bytes::BYTES_ID),
-                    body: $buffer->flush(),
-                    consumer: $this instanceof Consumer ? $this : throw new NsqException('what?'),
-                ),
-                default => throw new NsqException('Unexpected frame type: '.$type)
-            };
-
-            if ($frame instanceof Response && $frame->isHeartBeat()) {
-                yield $this->command('NOP');
-
-                return $this->readFrame();
-            }
-
-            return $frame;
-        });
-    }
-
-    /**
-     * @return Promise<void>
-     */
-    protected function checkIsOK(): Promise
-    {
-        return call(function (): \Generator {
-            /** @var Response $response */
-            $response = yield $this->readResponse();
-
-            if (!$response->isOk()) {
-                throw new BadResponse($response);
-            }
-
-            $this->logger->debug('Ok checked.');
-
-            return call(static function (): void {});
-        });
-    }
-
-    /**
-     * @return Promise<Response>
-     */
-    private function readResponse(): Promise
-    {
-        return call(function (): \Generator {
-            $frame = yield $this->readFrame();
-
-            if ($frame instanceof Error) {
-                if ($frame->type->terminateConnection) {
+        if (ErrorType::terminable($error)) {
             $this->close();
+
+            throw $error->toException();
+        }
     }
 
-                throw new NsqError($frame);
+    /**
+     * @return Promise<Frame\Response>
+     */
+    private function response(Stream $stream, Buffer $buffer): Promise
+    {
+        return call(function () use ($stream, $buffer): \Generator {
+            while (true) {
+                $response = Parser::parse($buffer);
+
+                if (null === $response && null !== ($chunk = yield $stream->read())) {
+                    $buffer->append($chunk);
+
+                    continue;
                 }
 
-            if (!$frame instanceof Response) {
-                throw new NsqException('Unreachable statement.');
+                if (!$response instanceof Frame\Response) {
+                    throw new NsqException();
                 }
 
-            return $frame;
+                return $response;
+            }
         });
     }
 }

src/Consumer.php

@@ -6,12 +6,12 @@ namespace Nsq;
 
 use Amp\Failure;
 use Amp\Promise;
-use Amp\Success;
-use Nsq\Exception\NsqError;
-use Nsq\Exception\NsqException;
-use Nsq\Protocol\Error;
-use Nsq\Protocol\Message;
-use Nsq\Protocol\Response;
+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;
 
@@ -20,61 +20,111 @@ final class Consumer extends Connection
     private int $rdy = 0;
 
     /**
-     * @return Promise<void>
+     * @var callable
      */
-    public function listen(
+    private $onMessage;
+
+    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,
-    ): Promise {
-        return call(function () use ($topic, $channel, $onMessage): \Generator {
-            yield $this->command('SUB', [$topic, $channel]);
-            yield $this->checkIsOK();
+        ?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) {
+            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.'));
+            }
 
-            asyncCall(function () use ($onMessage): \Generator {
             yield $this->rdy(2500);
 
-                while ($message = yield $this->readMessage()) {
-                    $command = yield $onMessage($message);
+            /** @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());
 
-                    if (true === $command) {
                                     break;
                                 }
 
-                    if ($this->rdy < 1000) {
-                        yield $this->rdy(2500);
+                                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;
+                        }
                     }
                 }
 
-                return new Success();
+                $this->stream = new NullStream();
             });
         });
     }
 
-    /**
-     * @return Promise<Message>
-     */
-    public function readMessage(): Promise
-    {
-        return call(function (): \Generator {
-            $frame = yield $this->readFrame();
-
-            if ($frame instanceof Message) {
-                return $frame;
-            }
-
-            if ($frame instanceof Error) {
-                if ($frame->type->terminateConnection) {
-                    yield $this->close();
-                }
-
-                throw new NsqError($frame);
-            }
-
-            throw new NsqException('Unreachable statement.');
-        });
-    }
-
     /**
      * Update RDY state (indicate you are ready to receive N messages).
      *
@@ -83,12 +133,13 @@ final class Consumer extends Connection
     public function rdy(int $count): Promise
     {
         if ($this->rdy === $count) {
-            return call(static function (): void {});
+            return call(static function (): void {
+            });
         }
 
         $this->rdy = $count;
 
-        return $this->command('RDY', (string) $count);
+        return $this->stream->write(Command::rdy($count));
     }
 
     /**
@@ -100,12 +151,9 @@ final class Consumer extends Connection
      */
     public function fin(string $id): Promise
     {
-        $promise = $this->command('FIN', $id);
-        $promise->onResolve(function (): void {
         --$this->rdy;
-        });
 
-        return $promise;
+        return $this->stream->write(Command::fin($id));
     }
 
     /**
@@ -120,12 +168,9 @@ final class Consumer extends Connection
      */
     public function req(string $id, int $timeout): Promise
     {
-        $promise = $this->command('REQ', [$id, $timeout]);
-        $promise->onResolve(function (): void {
         --$this->rdy;
-        });
 
-        return $promise;
+        return $this->stream->write(Command::req($id, $timeout));
     }
 
     /**
@@ -137,6 +182,6 @@ final class Consumer extends Connection
      */
     public function touch(string $id): Promise
     {
-        return $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;
 
-    /**
-     * 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)
+    public static function terminable(Frame\Error $error): bool
     {
-        $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/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 NotConnected 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/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

@@ -5,55 +5,95 @@ declare(strict_types=1);
 namespace Nsq;
 
 use Amp\Promise;
-use PHPinnacle\Buffer\ByteBuffer;
+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
 {
-    /**
-     * @return Promise<void>
-     */
-    public function pub(string $topic, string $body): Promise
+    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
     {
-        return call(function () use ($topic, $body): \Generator {
-            yield $this->command('PUB', $topic, $body);
-            yield $this->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): Promise
+    public function publish(string $topic, string | array $body, int $delay = 0): Promise
     {
-        return call(function () use ($topic, $bodies): \Generator {
-            $buffer = new ByteBuffer();
-
-            $buffer->appendUint32(\count($bodies));
-
+        if (0 < $delay) {
+            return call(
+                function (array $bodies) use ($topic, $delay): \Generator {
                     foreach ($bodies as $body) {
-                $buffer->appendUint32(\strlen($body));
-                $buffer->append($body);
+                        yield $this->stream->write(Command::dpub($topic, $body, $delay));
+                    }
+                },
+                (array) $body,
+            );
         }
 
-            yield $this->command('MPUB', $topic, $buffer->flush());
-            yield $this->checkIsOK();
-        });
+        $command = \is_array($body)
+            ? Command::mpub($topic, $body)
+            : Command::pub($topic, $body);
+
+        return $this->stream->write($command);
     }
 
-    /**
-     * @return Promise<void>
-     */
-    public function dpub(string $topic, string $body, int $delay): Promise
+    private function run(): void
     {
-        return call(function () use ($topic, $body, $delay): \Generator {
-            yield $this->command('DPUB', [$topic, $delay], $body);
-            yield $this->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,101 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Protocol;
-
-use Amp\Failure;
-use Amp\Promise;
-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;
-    }
-
-    /**
-     * @return Promise<void>
-     */
-    public function finish(): Promise
-    {
-        if ($this->finished) {
-            return new Failure(MessageAlreadyFinished::finish($this));
-        }
-
-        $this->finished = true;
-
-        return $this->consumer->fin($this->id);
-    }
-
-    /**
-     * @return Promise<void>
-     */
-    public function requeue(int $timeout): Promise
-    {
-        if ($this->finished) {
-            return new Failure(MessageAlreadyFinished::requeue($this));
-        }
-
-        $this->finished = true;
-
-        return $this->consumer->req($this->id, $timeout);
-    }
-
-    /**
-     * @return Promise<void>
-     */
-    public function touch(): Promise
-    {
-        if ($this->finished) {
-            return new Failure(MessageAlreadyFinished::touch($this));
-        }
-
-        return $this->consumer->touch($this->id);
-    }
-}

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/NsqInputStream.php

@@ -1,57 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\ByteStream\InputStream;
-use Amp\Promise;
-use Nsq\Bytes;
-use Nsq\Exception\NotConnected;
-use PHPinnacle\Buffer\ByteBuffer;
-use function Amp\call;
-
-final class NsqInputStream implements InputStream
-{
-    private ByteBuffer $buffer;
-
-    public function __construct(
-        private InputStream $inputStream,
-    ) {
-        $this->buffer = new ByteBuffer();
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function read(): Promise
-    {
-        return call(function (): \Generator {
-            $buffer = $this->buffer;
-
-            while ($buffer->size() < Bytes::BYTES_SIZE) {
-                $bytes = yield $this->inputStream->read();
-
-                if (null === $bytes) {
-                    throw new NotConnected();
-                }
-
-                $buffer->append($bytes);
-            }
-
-            $size = $buffer->consumeUint32();
-
-            while ($buffer->size() < $size) {
-                $bytes = yield $this->inputStream->read();
-
-                if (null === $bytes) {
-                    throw new NotConnected();
-                }
-
-                $buffer->append($bytes);
-            }
-
-            return $buffer->consume($size);
-        });
-    }
-}

src/Stream/NullStream.php

@@ -4,39 +4,34 @@ declare(strict_types=1);
 
 namespace Nsq\Stream;
 
-use Amp\ByteStream\InputStream;
-use Amp\ByteStream\OutputStream;
-use Amp\Failure;
 use Amp\Promise;
-use Nsq\Exception\NotConnected;
+use Amp\Success;
+use Nsq\Stream;
+use function Amp\call;
 
-final class NullStream implements InputStream, OutputStream
+final class NullStream implements Stream
 {
     /**
-     * {@inheritDoc}
+     * {@inheritdoc}
      */
     public function read(): Promise
     {
-        return new Failure(new NotConnected());
+        return new Success(null);
     }
 
     /**
-     * {@inheritDoc}
-     *
-     * @return Promise<void>
+     * {@inheritdoc}
      */
     public function write(string $data): Promise
     {
-        return new Failure(new NotConnected());
+        return call(static function (): void {
+        });
     }
 
     /**
-     * {@inheritDoc}
-     *
-     * @return Promise<void>
+     * {@inheritdoc}
      */
-    public function end(string $finalData = ''): Promise
+    public function close(): void
     {
-        return new Failure(new NotConnected());
     }
 }

src/Stream/SnappyInputStream.php

@@ -1,106 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\ByteStream\InputStream;
-use Amp\Promise;
-use Nsq\Exception\NotConnected;
-use PHPinnacle\Buffer\ByteBuffer;
-use Psr\Log\LoggerInterface;
-use function Amp\call;
-
-final class SnappyInputStream implements InputStream
-{
-    private ByteBuffer $buffer;
-
-    public function __construct(
-        private InputStream $inputStream,
-        private LoggerInterface $logger,
-    ) {
-        if (!\function_exists('snappy_uncompress')) {
-            throw new \LogicException('Snappy extension not installed.');
-        }
-
-        $this->buffer = new ByteBuffer();
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public function read(): Promise
-    {
-        return call(function (): \Generator {
-            $buffer = $this->buffer;
-
-            while ($buffer->size() < 4) {
-                $bytes = yield $this->inputStream->read();
-
-                if (null === $bytes) {
-                    throw new NotConnected();
-                }
-
-                $buffer->append($bytes);
-            }
-
-            /** @phpstan-ignore-next-line */
-            $chunkType = unpack('V', $buffer->consume(4))[1];
-
-            $size = $chunkType >> 8;
-            $chunkType &= 0xff;
-
-            $this->logger->debug('Snappy receive chunk [{chunk}], size [{size}]', [
-                'chunk' => $chunkType,
-                'size' => $size,
-            ]);
-
-            while ($buffer->size() < $size) {
-                $bytes = yield $this->inputStream->read();
-
-                if (null === $bytes) {
-                    throw new NotConnected();
-                }
-
-                $buffer->append($bytes);
-            }
-
-            switch ($chunkType) {
-                case 0xff:
-                    $this->logger->debug('Snappy identifier chunk');
-
-                    $buffer->discard(6); // discard identifier body
-
-                break;
-                case 0x00: // 'compressed',
-                    $this->logger->debug('Snappy compressed chunk');
-
-                    $data = $buffer
-                        ->discard(4) // discard checksum
-                        ->consume($size)
-                    ;
-
-                    $this->logger->debug('Snappy compressed data [{data}]', ['data' => $data]);
-
-                    return snappy_uncompress($data);
-                case 0x01: // 'uncompressed',
-                    $this->logger->debug('Snappy uncompressed chunk');
-
-                    $data = $buffer
-                        ->discard(4) // discard checksum
-                        ->consume($size)
-                    ;
-
-                    $this->logger->debug('Snappy uncompressed data [{data}]', ['data' => $data]);
-
-                    return $data;
-                case 0xfe:// 'padding',
-                    $this->logger->debug('Snappy padding chunk');
-
-                    $buffer->discard($size); // TODO ?
-            }
-
-            return $this->read();
-        });
-    }
-}

src/Stream/SnappyOutputStream.php

@@ -1,74 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Stream;
-
-use Amp\ByteStream\OutputStream;
-use Amp\Promise;
-use PHPinnacle\Buffer\ByteBuffer;
-
-final class SnappyOutputStream implements OutputStream
-{
-    private ByteBuffer $buffer;
-
-    public function __construct(
-        private OutputStream $outputStream,
-    ) {
-        if (!\function_exists('snappy_compress')) {
-            throw new \LogicException('Snappy extension not installed.');
-        }
-
-        $this->buffer = new ByteBuffer();
-    }
-
-    /**
-     * {@inheritDoc}
-     *
-     * @return Promise<void>
-     */
-    public function write(string $data): Promise
-    {
-        $identifierFrame = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
-        $compressedFrame = 0x00;
-        $uncompressedFrame = 0x01; // 11
-        $maxChunkLength = 65536;
-
-        $buffer = $this->buffer;
-        foreach ($identifierFrame as $bite) {
-            $buffer->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;
-
-            $buffer->append(pack('V', $chunkType + $size));
-            $buffer->append(pack('V', $maskedChecksum));
-            $buffer->append($chunk);
-        }
-
-        return $this->outputStream->write($buffer->flush());
-    }
-
-    /**
-     * {@inheritDoc}
-     *
-     * @return Promise<void>
-     */
-    public function end(string $finalData = ''): Promise
-    {
-        return $this->outputStream->end($finalData);
-    }
-}

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/MessageTest.php

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

tests/NsqTest.php

@@ -1,39 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-use Nsq\Config\ClientConfig;
-use PHPUnit\Framework\TestCase;
-
-final class NsqTest extends TestCase
-{
-    /**
-     * @dataProvider configs
-     */
-    public function test(ClientConfig $clientConfig): void
-    {
-        self::markTestSkipped('');
-    }
-
-    /**
-     * @return Generator<string, array<int, ClientConfig>>
-     */
-    public function configs(): Generator
-    {
-        yield 'default' => [
-            new ClientConfig(
-                heartbeatInterval: 3000,
-                snappy: false,
-                readTimeout: 1,
-            ),
-        ];
-
-        yield 'snappy' => [
-            new ClientConfig(
-                heartbeatInterval: 3000,
-                snappy: true,
-                readTimeout: 1,
-            ),
-        ];
-    }
-}

tests/ProducerTest.php

@@ -2,25 +2,60 @@
 
 declare(strict_types=1);
 
-use Nsq\Exception\NsqError;
+use Amp\Loop;
+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(NsqError::class);
+        $this->expectException(ServerException::class);
         $this->expectExceptionMessage($exceptionMessage);
 
-        $producer = new Producer('tcp://localhost:4150');
+        $producer = Producer::create('tcp://localhost:4150');
 
-        wait($producer->connect());
-        wait($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];
-        }
-    }
-}