{"record":{"id":"cf18b0d50760febd","repo":"php-amqplib/php-amqplib","slug":"broken-pipe-or-closed-connection-streamio","errorCode":null,"errorMessage":"Broken pipe or closed connection","messagePattern":"Broken pipe or closed connection","errorType":"exception","errorClass":"AMQPConnectionClosedException","httpStatus":null,"severity":"critical","filePath":"PhpAmqpLib/Wire/IO/StreamIO.php","lineNumber":187,"sourceCode":"    }\n\n    /**\n     * @inheritdoc\n     */\n    public function read($len)\n    {\n        $this->check_heartbeat();\n\n        list($timeout_sec, $timeout_uSec) = MiscHelper::splitSecondsMicroseconds($this->read_timeout);\n\n        $read_start = microtime(true);\n        $read = 0;\n        $data = '';\n\n        while ($read < $len) {\n            if (!is_resource($this->sock) || feof($this->sock)) {\n                $this->close();\n                throw new AMQPConnectionClosedException('Broken pipe or closed connection');\n            }\n\n            $this->setErrorHandler();\n            try {\n                $buffer = fread($this->sock, ($len - $read));\n                $this->throwOnError();\n            } catch (\\ErrorException $e) {\n                throw new AMQPDataReadException($e->getMessage(), $e->getCode(), $e->getPrevious());\n            } finally {\n                $this->restoreErrorHandler();\n            }\n\n            if ($buffer === false) {\n                throw new AMQPDataReadException('Error receiving data');\n            }\n\n            if ($buffer === '') {\n                $read_now = microtime(true);","sourceCodeStart":169,"sourceCodeEnd":205,"githubUrl":"https://github.com/php-amqplib/php-amqplib/blob/381b6f7c600e0e0c7463cdd7f7a1a3bc6268e5fd/PhpAmqpLib/Wire/IO/StreamIO.php#L169-L205","documentation":"StreamIO::read() detects that the socket is no longer a valid resource or has hit EOF before the requested bytes could be read. The connection is closed and AMQPConnectionClosedException('Broken pipe or closed connection') is thrown, signaling the AMQP connection is dead.","triggerScenarios":"Broker closed the TCP connection or the network dropped while StreamIO::read() was awaiting len bytes; is_resource($this->sock) fails or feof($this->sock) is true inside the read loop.","commonSituations":"RabbitMQ broker restart/crash, heartbeat timeout closing an idle connection, network interruption, firewall/LB killing idle TCP sessions, consumer reading after channel/connection was closed server-side.","solutions":["Check broker logs and connectivity (is RabbitMQ up? correct host/port/firewall rules?)","Configure heartbeats so dead connections are detected and handled predictably (heartbeat in AMQPConnectionConfig)","Wrap consumption/publishing in a try-catch for AMQPConnectionClosedException and reconnect with backoff","Investigate why the server closed: heartbeat timeouts, connection_limits, per-user channel limits, or idle LB timeouts"],"exampleFix":"// before\n$msg = $channel->basic_get($queue); // throws on dead connection\n// after\ntry {\n    $msg = $channel->basic_get($queue);\n} catch (AMQPConnectionClosedException $e) {\n    $connection = reconnectWithBackoff();\n    $channel = $connection->channel();\n    $msg = $channel->basic_get($queue);\n}","handlingStrategy":"try-catch","validationCode":"// before consuming/publishing, verify the broker is reachable\nif (@fsockopen($host, $port, $errno, $errstr, 2) === false) {\n    throw new \\RuntimeException(\"Broker unreachable: $errstr ($errno)\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    $channel->basic_consume($queue, ...);\n} catch (PhpAmqpLib\\Exception\\AMQPConnectionClosedException $e) {\n    $this->connection = $this->factory->createConnectionWithBackoff();\n    $this->channel = $this->connection->channel();\n    $this->resubscribe();\n}","preventionTips":["Enable heartbeats (e.g. 15-60s) so dead connections are detected promptly","Keep read/write timeouts realistic for your operations","Supervise long-running consumers with reconnect + resubscribe logic","Monitor broker logs for connection closes (timeouts, limits, alarms)"],"tags":["php","amqp","broken-pipe","network","connection-closed"],"backgroundTag":"broken-pipe","analyzedSha":"381b6f7c600e0e0c7463cdd7f7a1a3bc6268e5fd","analyzedAt":"2026-09-20T11:37:12.097Z","contentChangedAt":"2026-09-20T11:37:12.097Z","schemaVersion":2},"datasetVersion":"2026-09-21T04:17:39.646Z"}