Consumer tag assignment fix; timeout consistency (#7)

Author: mmasiukevich

src/Channel.php

@@ -224,7 +224,9 @@ public function consume
                 /** @var Protocol\BasicConsumeOkFrame $result */
                 $result = yield $this->await(Protocol\BasicConsumeOkFrame::class);
 
-                $consumerTag = $result->consumerTag;
+                if('' === $consumerTag) {
+                    $consumerTag = $result->consumerTag;
+                }
             }
 
             $this->startConsuming();

src/Client.php

@@ -83,8 +83,10 @@ public function connect(): Promise
 
             $this->connection = new Connection($this->config->uri());
 
+            $timeout = $this->config->timeout() * 1000;
+
             yield $this->connection->open(
-                $this->config->timeout(),
+                $timeout,
                 $this->config->tcpAttempts(),
                 $this->config->tcpNoDelay()
             );

src/Connection.php

@@ -88,7 +88,7 @@ public function send(Protocol\AbstractFrame $frame): Promise
             // payload already supplied
         } elseif ($frame instanceof MethodFrame || $frame instanceof ContentHeaderFrame) {
             $buffer = $frame->pack();
-        
+
             $frame->size    = $buffer->size();
             $frame->payload = $buffer;
         } elseif ($frame instanceof Protocol\ContentBodyFrame) {
@@ -156,10 +156,11 @@ public function cancel(int $channel): void
     public function open(int $timeout, int $maxAttempts, bool $noDelay): Promise
     {
         return call(function () use ($timeout, $maxAttempts, $noDelay) {
-            $context = (new ClientConnectContext)
-                ->withConnectTimeout($timeout)
-                ->withMaxAttempts($maxAttempts)
-            ;
+            $context = (new ClientConnectContext)->withMaxAttempts($maxAttempts);
+
+            if($timeout > 0) {
+                $context = $context->withConnectTimeout($timeout);
+            }
 
             if ($noDelay) {
                 $context->withTcpNoDelay();
@@ -182,7 +183,7 @@ public function open(int $timeout, int $maxAttempts, bool $noDelay): Promise
                     }
                 }
 
-                unset($this->socket);
+                $this->socket = null;
             });
         });
     }
@@ -197,7 +198,7 @@ public function heartbeat(int $interval): void
         $milliseconds = $interval * 1000;
 
         $this->heartbeat = Loop::repeat($milliseconds, function($watcher) use ($milliseconds) {
-            if (!$this->socket) {
+            if (null === $this->socket) {
                 Loop::cancel($watcher);
 
                 return;