Merge branch 'master' of github.com:hyperf/hyperf

CHANGELOG.md

@@ -1,5 +1,9 @@
 # v1.1.15 - TBD
 
+## Fixed
+
+- [#1258](https://github.com/hyperf/hyperf/pull/1258) Fixed CRITICAL error that socket of process is unavailable when amqp send heartbeat failed.
+
 # v1.1.14 - 2010-01-10
 
 ## Added
@@ -13,7 +17,7 @@
 ## Fixed
 
 - [#1208](https://github.com/hyperf/hyperf/pull/1208) Fixed bug that exception and error cannot be resolved successfully in TcpServer.
-- [#1208](https://github.com/hyperf/hyperf/pull/1208) Fixed bug that request id of json-rpc MUST is equal to response id.
+- [#1208](https://github.com/hyperf/hyperf/pull/1208) Fixed bug that json-rpc has not validated the request id whether is equal to response id.
 - [#1223](https://github.com/hyperf/hyperf/pull/1223) Fixed the scanner will missing the packages at require-dev of composer.json
 - [#1254](https://github.com/hyperf/hyperf/pull/1254) Fixed bash not found on some environment like Alpine when execute `init-proxy.sh`.
 

src/amqp/src/Connection/Socket.php

@@ -12,8 +12,11 @@ declare(strict_types=1);
 
 namespace Hyperf\Amqp\Connection;
 
+use Hyperf\Contract\StdoutLoggerInterface;
+use Hyperf\Utils\ApplicationContext;
 use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Wire\AMQPWriter;
+use Psr\Log\LoggerInterface;
 use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\Client;
 use Swoole\Timer;
@@ -82,6 +85,9 @@ class Socket
         }
 
         $client = $this->channel->pop($this->waitTimeout);
+        if ($client === false) {
+            throw new AMQPRuntimeException('Socket of keepaliveIO is exhausted. Cannot establish new socket before wait_timeout.');
+        }
 
         $result = $closure($client);
 
@@ -146,7 +152,15 @@ class Socket
     {
         $this->clear();
         $this->timerId = Timer::tick($this->heartbeat * 1000, function () {
-            $this->heartbeat();
+            try {
+                $this->heartbeat();
+            } catch (\Throwable $throwable) {
+                $this->close();
+                if ($logger = $this->getLogger()) {
+                    $message = sprintf('KeepaliveIO heartbeat failed, %s', $throwable->getMessage());
+                    $logger->error($message);
+                }
+            }
         });
     }
 
@@ -157,4 +171,18 @@ class Socket
             $this->timerId = null;
         }
     }
+
+    protected function getLogger(): ?LoggerInterface
+    {
+        if (! ApplicationContext::hasContainer()) {
+            return null;
+        }
+
+        $container = ApplicationContext::getContainer();
+        if (! $container->has(StdoutLoggerInterface::class)) {
+            return null;
+        }
+
+        return $container->get(StdoutLoggerInterface::class);
+    }
 }