diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 99d86a1..3194617 100644
--- a/.github/workflows/ci.yaml
+++ b/.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
 
diff --git a/.php_cs.dist b/.php_cs.dist
index 84cf4b1..7730898 100644
--- a/.php_cs.dist
+++ b/.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'],
         ],
diff --git a/README.md b/README.md
index cf52be2..95a554a 100644
--- a/README.md
+++ b/README.md
@@ -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->defer('topic', 'Deferred message', delay: 5000);
 ```
 
 ### Consumer
 
 ```php
 use Nsq\Consumer;
-use Nsq\Protocol\Message;
+use Nsq\Message;
 
-$consumer = new Consumer(
-    topic: 'topic', 
+$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
diff --git a/composer.json b/composer.json
index e0a5d61..2c64aff 100644
--- a/composer.json
+++ b/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",
diff --git a/docker-compose.yml b/docker-compose.yml
index 324e1e2..8bdd629 100644
--- a/docker-compose.yml
+++ b/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
diff --git a/examples/consumer.php b/examples/consumer.php
new file mode 100644
index 0000000..d946fb1
--- /dev/null
+++ b/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();
+});
diff --git a/examples/producer.php b/examples/producer.php
new file mode 100644
index 0000000..2674d6a
--- /dev/null
+++ b/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!'));
+    }
+});
diff --git a/examples/reader.php b/examples/reader.php
new file mode 100644
index 0000000..68dddee
--- /dev/null
+++ b/examples/reader.php
@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+require __DIR__.'/../vendor/autoload.php';
+
+use Amp\ByteStream;
+use Amp\Log\ConsoleFormatter;
+use Amp\Log\StreamHandler;
+use Monolog\Logger;
+use Monolog\Processor\PsrLogMessageProcessor;
+use Nsq\Config\ClientConfig;
+use Nsq\Reader;
+use function Amp\Promise\wait;
+
+$handler = new StreamHandler(ByteStream\getStdout());
+$handler->setFormatter(new ConsoleFormatter());
+$logger = new Logger('publisher', [$handler], [new PsrLogMessageProcessor()]);
+
+$reader = new Reader(
+    'tcp://localhost:4150',
+    topic: 'local',
+    channel: 'local',
+    clientConfig: new ClientConfig(
+        deflate: false,
+        snappy: false,
+    ),
+    logger: $logger,
+);
+
+wait($reader->connect());
+
+while (true) {
+    $message = wait($reader->consume());
+
+    $logger->info('Received: {body}', ['body' => $message->body]);
+
+    wait($message->finish());
+}
diff --git a/src/Buffer.php b/src/Buffer.php
index 7cfad1c..49784ef 100644
--- a/src/Buffer.php
+++ b/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 __construct(string $initial = '')
+    public function readUInt32LE(): int
     {
-        $this->buffer = new ByteBuffer($initial);
-    }
-
-    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();
+        /** @phpstan-ignore-next-line  */
+        return unpack('V', $this->consume(4))[1];
     }
 
     public function consumeTimestamp(): int
     {
-        /** @see Bytes::BYTES_TIMESTAMP */
-        return $this->buffer->consumeInt64();
+        return $this->consumeUint64();
     }
 
     public function consumeAttempts(): int
     {
-        /** @see Bytes::BYTES_ATTEMPTS */
-        return $this->buffer->consumeUint16();
+        return $this->consumeUint16();
     }
 
-    public function consumeId(): string
+    public function consumeMessageID(): string
     {
-        return $this->buffer->consume(Bytes::BYTES_ID);
-    }
-
-    public function size(): int
-    {
-        return $this->buffer->size();
-    }
-
-    public function bytes(): string
-    {
-        return $this->buffer->bytes();
-    }
-
-    public function flush(): string
-    {
-        return $this->buffer->flush();
+        return $this->consume(16);
     }
 }
diff --git a/src/Command.php b/src/Command.php
new file mode 100644
index 0000000..17a951d
--- /dev/null
+++ b/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();
+    }
+}
diff --git a/src/Connection.php b/src/Connection.php
index 1f8bc9a..1bd2778 100644
--- a/src/Connection.php
+++ b/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\Protocol\Frame;
-use Nsq\Protocol\Message;
-use Nsq\Protocol\Response;
-use Nsq\Socket\DeflateSocket;
-use Nsq\Socket\RawSocket;
-use Nsq\Socket\SnappySocket;
-use Psr\Log\LoggerAwareTrait;
+use Nsq\Frame\Response;
+use Nsq\Stream\GzipStream;
+use Nsq\Stream\NullStream;
+use Nsq\Stream\SnappyStream;
+use Nsq\Stream\SocketStream;
 use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
+use function Amp\call;
 
 /**
  * @internal
  */
 abstract class Connection
 {
-    use LoggerAwareTrait;
-
-    protected ClientConfig $clientConfig;
-
-    private NsqSocket $socket;
-
-    private ConnectionConfig $connectionConfig;
-
-    private bool $closed = false;
+    protected Stream $stream;
 
     public function __construct(
         private string $address,
-        ClientConfig $clientConfig = null,
-        LoggerInterface $logger = null,
+        private ClientConfig $clientConfig,
+        private LoggerInterface $logger,
     ) {
-        $this->logger = $logger ?? new NullLogger();
-        $this->clientConfig = $clientConfig ?? new ClientConfig();
+        $this->stream = new NullStream();
+    }
 
-        $socket = new RawSocket($this->address, $this->logger);
-        $socket->write('  V2');
-
-        $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));
-        }
+    public function __destruct()
+    {
+        $this->close();
     }
 
     /**
-     * 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) {
-            return;
-        }
+//        $this->stream->write(Command::cls());
 
-        try {
-            $this->command('CLS');
-            $this->socket->close();
-        } catch (\Throwable $e) {
-        }
-
-        $this->closed = true;
+        $this->stream->close();
+        $this->stream = new NullStream();
     }
 
-    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) {
-            throw new NotConnected('Connection closed.');
-        }
+        return call(function () use ($stream, $buffer): \Generator {
+            while (true) {
+                $response = Parser::parse($buffer);
 
-        $command = [] === $params
-            ? $command
-            : implode(' ', [$command, ...((array) $params)]);
+                if (null === $response && null !== ($chunk = yield $stream->read())) {
+                    $buffer->append($chunk);
 
-        $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;
-    }
-
-    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();
+                return $response;
             }
-
-            throw new NsqError($frame);
-        }
-
-        throw new NsqException('Unreachable statement.');
+        });
     }
 }
diff --git a/src/Consumer.php b/src/Consumer.php
index 0b1d4c1..bf93f69 100644
--- a/src/Consumer.php
+++ b/src/Consumer.php
@@ -4,108 +4,148 @@ 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\NsqException;
-use Nsq\Protocol\Error;
-use Nsq\Protocol\Message;
-use Nsq\Protocol\Response;
+use Nsq\Exception\ConsumerException;
+use Nsq\Frame\Response;
 use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use function Amp\asyncCall;
+use function Amp\call;
 
-final class Consumer extends Connection
+final class Consumer extends Connection implements ConsumerInterface
 {
     private int $rdy = 0;
 
+    /**
+     * @var callable
+     */
+    private $onMessage;
+
     public function __construct(
+        private string $address,
         private string $topic,
         private string $channel,
-        string $address,
-        ClientConfig $clientConfig = null,
-        LoggerInterface $logger = null
+        callable $onMessage,
+        ClientConfig $clientConfig,
+        private LoggerInterface $logger,
     ) {
-        parent::__construct($address, $clientConfig, $logger);
+        parent::__construct(
+            $this->address,
+            $clientConfig,
+            $this->logger,
+        );
+
+        $this->onMessage = $onMessage;
     }
 
-    /**
-     * @psalm-return Generator<int, Message|float|null, int|null, void>
-     */
-    public function generator(): \Generator
-    {
-        $this->command('SUB', [$this->topic, $this->channel])->checkIsOK();
-
-        while (true) {
-            $this->rdy(1);
-
-            $timeout = $this->clientConfig->readTimeout;
-
-            do {
-                $deadline = microtime(true) + $timeout;
-
-                $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 static function create(
+        string $address,
+        string $topic,
+        string $channel,
+        callable $onMessage,
+        ?ClientConfig $clientConfig = null,
+        ?LoggerInterface $logger = null,
+    ): self {
+        return new self(
+            $address,
+            $topic,
+            $channel,
+            $onMessage,
+            $clientConfig ?? new ClientConfig(),
+            $logger ?? new NullLogger(),
+        );
     }
 
-    public function readMessage(): ?Message
+    public function connect(): Promise
     {
-        $frame = $this->readFrame();
+        return call(function (): \Generator {
+            yield parent::connect();
 
-        if ($frame instanceof Message) {
-            return $frame;
-        }
+            $this->run();
+        });
+    }
 
-        if ($frame instanceof Response && $frame->isHeartBeat()) {
-            $this->command('NOP');
+    private function run(): void
+    {
+        $buffer = new Buffer();
 
-            return null;
-        }
+        asyncCall(function () use ($buffer): \Generator {
+            yield $this->stream->write(Command::sub($this->topic, $this->channel));
 
-        if ($frame instanceof Error) {
-            if ($frame->type->terminateConnection) {
-                $this->close();
+            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;
+                        }
+                    }
+                }
+            });
+        });
     }
 
     /**
      * 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 +154,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));
     }
 }
diff --git a/src/ConsumerInterface.php b/src/ConsumerInterface.php
new file mode 100644
index 0000000..672adcf
--- /dev/null
+++ b/src/ConsumerInterface.php
@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq;
+
+use Amp\Promise;
+
+interface ConsumerInterface
+{
+    /**
+     * Update RDY state (indicate you are ready to receive N messages).
+     *
+     * @return Promise<void>
+     */
+    public function rdy(int $count): Promise;
+
+    /**
+     * Finish a message (indicate successful processing).
+     *
+     * @return Promise<void>
+     *
+     * @internal
+     */
+    public function fin(string $id): Promise;
+
+    /**
+     * Re-queue a message (indicate failure to process) The re-queued message is placed at the tail of the queue,
+     * equivalent to having just published it, but for various implementation specific reasons that behavior should not
+     * 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): Promise;
+
+    /**
+     * Reset the timeout for an in-flight message.
+     *
+     * @return Promise<void>
+     *
+     * @internal
+     */
+    public function touch(string $id): Promise;
+}
diff --git a/src/Protocol/ErrorType.php b/src/ErrorType.php
similarity index 92%
rename from src/Protocol/ErrorType.php
rename to src/ErrorType.php
index 828afc5..1226ad6 100644
--- a/src/Protocol/ErrorType.php
+++ b/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;
     }
 }
diff --git a/src/Exception/BadResponse.php b/src/Exception/BadResponse.php
deleted file mode 100644
index 6be3633..0000000
--- a/src/Exception/BadResponse.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Exception/ConnectionFail.php b/src/Exception/ConnectionFail.php
deleted file mode 100644
index 6920e99..0000000
--- a/src/Exception/ConnectionFail.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Exception/ConsumerException.php b/src/Exception/ConsumerException.php
new file mode 100644
index 0000000..f184e53
--- /dev/null
+++ b/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));
+    }
+}
diff --git a/src/Exception/MessageAlreadyFinished.php b/src/Exception/MessageAlreadyFinished.php
deleted file mode 100644
index b50bbcc..0000000
--- a/src/Exception/MessageAlreadyFinished.php
+++ /dev/null
@@ -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.');
-    }
-}
diff --git a/src/Exception/MessageException.php b/src/Exception/MessageException.php
new file mode 100644
index 0000000..8906b0e
--- /dev/null
+++ b/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));
+    }
+}
diff --git a/src/Exception/NotConnected.php b/src/Exception/NotConnected.php
deleted file mode 100644
index e184835..0000000
--- a/src/Exception/NotConnected.php
+++ /dev/null
@@ -1,9 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Exception;
-
-final class NotConnected extends NsqException
-{
-}
diff --git a/src/Exception/NsqError.php b/src/Exception/NsqError.php
deleted file mode 100644
index 44cc06d..0000000
--- a/src/Exception/NsqError.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Exception/NullReceived.php b/src/Exception/ServerException.php
similarity index 56%
rename from src/Exception/NullReceived.php
rename to src/Exception/ServerException.php
index 8ac0055..32659ea 100644
--- a/src/Exception/NullReceived.php
+++ b/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
 {
 }
diff --git a/src/Exception/SnappyException.php b/src/Exception/SnappyException.php
new file mode 100644
index 0000000..ba22b59
--- /dev/null
+++ b/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.');
+    }
+}
diff --git a/src/Frame.php b/src/Frame.php
new file mode 100644
index 0000000..30a3618
--- /dev/null
+++ b/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;
+    }
+}
diff --git a/src/Frame/Error.php b/src/Frame/Error.php
new file mode 100644
index 0000000..d946dbf
--- /dev/null
+++ b/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);
+    }
+}
diff --git a/src/Frame/Message.php b/src/Frame/Message.php
new file mode 100644
index 0000000..b199468
--- /dev/null
+++ b/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);
+    }
+}
diff --git a/src/Protocol/Response.php b/src/Frame/Response.php
similarity index 54%
rename from src/Protocol/Response.php
rename to src/Frame/Response.php
index 9850e45..51afb6d 100644
--- a/src/Protocol/Response.php
+++ b/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);
     }
 }
diff --git a/src/Message.php b/src/Message.php
new file mode 100644
index 0000000..8ec1104
--- /dev/null
+++ b/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 ConsumerInterface $consumer,
+    ) {
+    }
+
+    public static function compose(Frame\Message $message, ConsumerInterface $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;
+        });
+    }
+}
diff --git a/src/NsqSocket.php b/src/NsqSocket.php
deleted file mode 100644
index cb6966a..0000000
--- a/src/NsqSocket.php
+++ /dev/null
@@ -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) {
-        }
-    }
-}
diff --git a/src/Parser.php b/src/Parser.php
new file mode 100644
index 0000000..0d3e948
--- /dev/null
+++ b/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)),
+        };
+    }
+}
diff --git a/src/Producer.php b/src/Producer.php
index 6eb7b0e..48f62f1 100644
--- a/src/Producer.php
+++ b/src/Producer.php
@@ -4,38 +4,85 @@ declare(strict_types=1);
 
 namespace Nsq;
 
-use PHPinnacle\Buffer\ByteBuffer;
+use Amp\Promise;
+use Nsq\Config\ClientConfig;
+use Nsq\Exception\NsqException;
+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();
+        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): Promise
     {
-        static $buffer;
-        $buffer ??= new ByteBuffer();
+        $command = \is_array($body)
+            ? Command::mpub($topic, $body)
+            : Command::pub($topic, $body);
 
-        $buffer->appendUint32(\count($bodies));
-
-        foreach ($bodies as $body) {
-            $buffer->appendUint32(\strlen($body));
-            $buffer->append($body);
-        }
-
-        $this->command('MPUB', $topic, $buffer->flush())->checkIsOK();
+        return $this->stream->write($command);
     }
 
-    public function dpub(string $topic, string $body, int $delay): void
+    /**
+     * @return Promise<void>
+     */
+    public function defer(string $topic, string $body, int $delay): Promise
     {
-        $this->command('DPUB', [$topic, $delay], $body)->checkIsOK();
+        return $this->stream->write(Command::dpub($topic, $body, $delay));
+    }
+
+    private function run(): void
+    {
+        $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.');
+                    }
+                }
+            }
+        });
     }
 }
diff --git a/src/Protocol/Error.php b/src/Protocol/Error.php
deleted file mode 100644
index ae789e6..0000000
--- a/src/Protocol/Error.php
+++ /dev/null
@@ -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]);
-    }
-}
diff --git a/src/Protocol/Frame.php b/src/Protocol/Frame.php
deleted file mode 100644
index b366a7c..0000000
--- a/src/Protocol/Frame.php
+++ /dev/null
@@ -1,16 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Protocol;
-
-abstract class Frame
-{
-    public function __construct(
-        /**
-         * @psalm-readonly
-         */
-        public int $length,
-    ) {
-    }
-}
diff --git a/src/Protocol/Message.php b/src/Protocol/Message.php
deleted file mode 100644
index be8a68f..0000000
--- a/src/Protocol/Message.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Reader.php b/src/Reader.php
new file mode 100644
index 0000000..5b4e0e7
--- /dev/null
+++ b/src/Reader.php
@@ -0,0 +1,214 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq;
+
+use Amp\Deferred;
+use Amp\Promise;
+use Amp\Success;
+use Nsq\Config\ClientConfig;
+use Nsq\Exception\ConsumerException;
+use Nsq\Frame\Response;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use function Amp\asyncCall;
+use function Amp\call;
+
+final class Reader extends Connection implements ConsumerInterface
+{
+    private int $rdy = 0;
+
+    /**
+     * @var array<int, Deferred<Message>>
+     */
+    private array $deferreds = [];
+
+    /**
+     * @var array<int, Message>
+     */
+    private array $messages = [];
+
+    public function __construct(
+        private string $address,
+        private string $topic,
+        private string $channel,
+        ClientConfig $clientConfig,
+        private LoggerInterface $logger,
+    ) {
+        parent::__construct(
+            $this->address,
+            $clientConfig,
+            $this->logger,
+        );
+    }
+
+    public static function create(
+        string $address,
+        string $topic,
+        string $channel,
+        ?ClientConfig $clientConfig = null,
+        ?LoggerInterface $logger = null,
+    ): self {
+        return new self(
+            $address,
+            $topic,
+            $channel,
+            $clientConfig ?? new ClientConfig(),
+            $logger ?? new NullLogger(),
+        );
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function connect(): Promise
+    {
+        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()) {
+                throw new ConsumerException('Fail subscription.');
+            }
+
+            yield $this->rdy(1);
+
+            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);
+
+                                    $deferred = array_pop($this->deferreds);
+
+                                    if (null !== $deferred) {
+                                        $deferred->fail($frame->toException());
+                                    }
+
+                                    break;
+                                case $frame instanceof Frame\Message:
+                                    $message = Message::compose($frame, $this);
+
+                                    $deferred = array_pop($this->deferreds);
+
+                                    if (null === $deferred) {
+                                        $this->messages[] = $message;
+                                    } else {
+                                        $deferred->resolve($message);
+                                    }
+
+                                    break;
+                            }
+                        }
+                    }
+                }
+            );
+        });
+    }
+
+    /**
+     * @return Promise<Message>
+     */
+    public function consume(): Promise
+    {
+        $message = array_pop($this->messages);
+
+        if (null !== $message) {
+            return new Success($message);
+        }
+
+        $this->deferreds[] = $deferred = new Deferred();
+
+        return $deferred->promise();
+    }
+
+    /**
+     * Update RDY state (indicate you are ready to receive N messages).
+     *
+     * @return Promise<void>
+     */
+    public function rdy(int $count): Promise
+    {
+        if ($this->rdy === $count) {
+            return call(static function (): void {
+            });
+        }
+
+        $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): Promise
+    {
+        --$this->rdy;
+
+        return $this->stream->write(Command::fin($id));
+    }
+
+    /**
+     * Re-queue a message (indicate failure to process) The re-queued message is placed at the tail of the queue,
+     * equivalent to having just published it, but for various implementation specific reasons that behavior should not
+     * 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): Promise
+    {
+        --$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): Promise
+    {
+        return $this->stream->write(Command::touch($id));
+    }
+}
diff --git a/src/Reconnect/ExponentialStrategy.php b/src/Reconnect/ExponentialStrategy.php
deleted file mode 100644
index 2454977..0000000
--- a/src/Reconnect/ExponentialStrategy.php
+++ /dev/null
@@ -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;
-    }
-}
diff --git a/src/Reconnect/RealTimeProvider.php b/src/Reconnect/RealTimeProvider.php
deleted file mode 100644
index c984a47..0000000
--- a/src/Reconnect/RealTimeProvider.php
+++ /dev/null
@@ -1,13 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Reconnect;
-
-final class RealTimeProvider implements TimeProvider
-{
-    public function time(): int
-    {
-        return time();
-    }
-}
diff --git a/src/Reconnect/ReconnectStrategy.php b/src/Reconnect/ReconnectStrategy.php
deleted file mode 100644
index 7b8a041..0000000
--- a/src/Reconnect/ReconnectStrategy.php
+++ /dev/null
@@ -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;
-}
diff --git a/src/Reconnect/TimeProvider.php b/src/Reconnect/TimeProvider.php
deleted file mode 100644
index a603e20..0000000
--- a/src/Reconnect/TimeProvider.php
+++ /dev/null
@@ -1,10 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Nsq\Reconnect;
-
-interface TimeProvider
-{
-    public function time(): int;
-}
diff --git a/src/Socket/DeflateSocket.php b/src/Socket/DeflateSocket.php
deleted file mode 100644
index 8aadc3a..0000000
--- a/src/Socket/DeflateSocket.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Socket/RawSocket.php b/src/Socket/RawSocket.php
deleted file mode 100644
index e661682..0000000
--- a/src/Socket/RawSocket.php
+++ /dev/null
@@ -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
-    }
-}
diff --git a/src/Socket/SnappySocket.php b/src/Socket/SnappySocket.php
deleted file mode 100644
index ac22e34..0000000
--- a/src/Socket/SnappySocket.php
+++ /dev/null
@@ -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);
-    }
-}
diff --git a/src/Socket/Socket.php b/src/Socket/Socket.php
deleted file mode 100644
index 6ea29b3..0000000
--- a/src/Socket/Socket.php
+++ /dev/null
@@ -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;
-}
diff --git a/src/Stream.php b/src/Stream.php
new file mode 100644
index 0000000..177b551
--- /dev/null
+++ b/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;
+}
diff --git a/src/Stream/GzipStream.php b/src/Stream/GzipStream.php
new file mode 100644
index 0000000..5d2ed9f
--- /dev/null
+++ b/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();
+    }
+}
diff --git a/src/Stream/NullStream.php b/src/Stream/NullStream.php
new file mode 100644
index 0000000..cf95f04
--- /dev/null
+++ b/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
+    {
+    }
+}
diff --git a/src/Stream/SnappyStream.php b/src/Stream/SnappyStream.php
new file mode 100644
index 0000000..63d618f
--- /dev/null
+++ b/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;
+    }
+}
diff --git a/src/Stream/SocketStream.php b/src/Stream/SocketStream.php
new file mode 100644
index 0000000..2036f1e
--- /dev/null
+++ b/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();
+    }
+}
diff --git a/tests/ErrorTypeTest.php b/tests/ErrorTypeTest.php
new file mode 100644
index 0000000..88389b5
--- /dev/null
+++ b/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<int, Error, 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];
+    }
+}
diff --git a/tests/ExponentialStrategyTest.php b/tests/ExponentialStrategyTest.php
deleted file mode 100644
index cee0065..0000000
--- a/tests/ExponentialStrategyTest.php
+++ /dev/null
@@ -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;
-    }
-}
diff --git a/tests/MessageTest.php b/tests/MessageTest.php
index 7489018..8f58097 100644
--- a/tests/MessageTest.php
+++ b/tests/MessageTest.php
@@ -2,9 +2,11 @@
 
 declare(strict_types=1);
 
-use Nsq\Consumer;
-use Nsq\Exception\MessageAlreadyFinished;
-use Nsq\Protocol\Message;
+use Amp\Loop;
+use Amp\Success;
+use Nsq\ConsumerInterface;
+use Nsq\Exception\MessageException;
+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();
-
-        self::assertTrue($message->isFinished());
-
-        $this->expectException(MessageAlreadyFinished::class);
-        $this->expectExceptionMessage('Can\'t finish message as it already finished.');
-
-        $message->finish();
+        Loop::run(function () use ($message): Generator {
+            yield $message->finish();
+            yield $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);
-
-        self::assertTrue($message->isFinished());
-
-        $this->expectException(MessageAlreadyFinished::class);
-        $this->expectExceptionMessage('Can\'t requeue message as it already finished.');
-
-        $message->requeue(5);
+        Loop::run(function () use ($message): Generator {
+            yield $message->requeue(1);
+            yield $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();
-
-        $this->expectException(MessageAlreadyFinished::class);
-        $this->expectExceptionMessage('Can\'t touch message as it already finished.');
-
-        $message->touch();
+        Loop::run(function () use ($message): Generator {
+            yield $message->finish();
+            yield $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(ConsumerInterface::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)];
     }
 }
diff --git a/tests/NsqTest.php b/tests/NsqTest.php
index 0738a4b..0bded8e 100644
--- a/tests/NsqTest.php
+++ b/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');
-        $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());
+        self::markTestSkipped('');
     }
 
     /**
diff --git a/tests/ProducerTest.php b/tests/ProducerTest.php
index e7f728e..9c7d5d1 100644
--- a/tests/ProducerTest.php
+++ b/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->pub($topic, $body);
+        $producer = Producer::create('tcp://localhost:4150');
+
+        Loop::run(static function () use ($producer, $topic, $body): Generator {
+            yield $producer->connect();
+
+            yield $producer->publish($topic, $body);
+        });
     }
 
     /**
diff --git a/tests/Protocol/ErrorTypeTest.php b/tests/Protocol/ErrorTypeTest.php
deleted file mode 100644
index b9fdc83..0000000
--- a/tests/Protocol/ErrorTypeTest.php
+++ /dev/null
@@ -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];
-        }
-    }
-}