Merge branch '2.2'

src/AMQPConnection.php

@@ -69,6 +69,11 @@ class AMQPConnection extends AbstractConnection
      */
     protected $channelManager;
 
+    /**
+     * @var bool
+     */
+    protected $exited = false;
+
     public function __construct(
         string $user,
         string $password,
@@ -141,7 +146,7 @@ public function channel($channel_id = null)
         $this->channelManager->close($channel_id);
         $this->channelManager->get($channel_id, true);
 
-        return parent::channel($channel_id); // TODO: Change the autogenerated stub
+        return parent::channel($channel_id);
     }
 
     public function getConfirmChannel(): AMQPChannel
@@ -209,7 +214,8 @@ protected function loop(): void
                     $this->channelManager->get($channel)->push([$frame_type, $payload], 0.001);
                 }
             } catch (\Throwable $exception) {
-                $this->logger && $this->logger->error('Recv loop broken. The reason is ' . (string) $exception);
+                $level = $this->exited ? 'warning' : 'error';
+                $this->logger && $this->logger->log($level, 'Recv loop broken. The reason is ' . (string) $exception);
             } finally {
                 $this->loop = false;
                 $this->close();
@@ -239,6 +245,8 @@ protected function heartbeat(): void
             Coroutine::create(function () {
                 while (true) {
                     if (CoordinatorManager::until(Constants::WORKER_EXIT)->yield($this->getHeartbeat())) {
+                        $this->exited = true;
+                        $this->close();
                         break;
                     }