{"record":{"id":"26256d0d1ac30c1e","repo":"php-amqplib/php-amqplib","slug":"broken-pipe-or-closed-connection","errorCode":null,"errorMessage":"Broken pipe or closed connection","messagePattern":"Broken pipe or closed connection","errorType":"exception","errorClass":"AMQPConnectionClosedException","httpStatus":null,"severity":"critical","filePath":"PhpAmqpLib/Connection/AbstractConnection.php","lineNumber":591,"sourceCode":"        $this->debug->debug_method_signature1($method_sig);\n\n        return $pkt;\n    }\n\n    /**\n     * Waits for a frame from the server\n     *\n     * @param int|float|null $timeout\n     * @return Frame\n     * @throws \\Exception\n     * @throws \\PhpAmqpLib\\Exception\\AMQPTimeoutException\n     * @throws AMQPRuntimeException\n     */\n    protected function wait_frame($timeout = 0): Frame\n    {\n        if (null === $this->input) {\n            $this->setIsConnected(false);\n            throw new AMQPConnectionClosedException('Broken pipe or closed connection');\n        }\n\n        $currentTimeout = $this->input->getTimeout();\n        $this->input->setTimeout($timeout);\n\n        try {\n            $header = $this->input->readFrameHeader();\n            $frame_type = $header['type'];\n            if (!$this->constants->isFrameType($frame_type)) {\n                throw new AMQPInvalidFrameException('Invalid frame type ' . $frame_type);\n            }\n            $size = $header['size'];\n\n            // payload + ch\n            $result = unpack('a' . $size . 'payload/Cch', $this->input->read(AMQPReader::OCTET + $size));\n            $ch = $result['ch'];\n            $frame = new Frame($frame_type, $header['channel'], $size, $result['payload']);\n        } catch (AMQPTimeoutException $e) {","sourceCodeStart":573,"sourceCodeEnd":609,"githubUrl":"https://github.com/php-amqplib/php-amqplib/blob/381b6f7c600e0e0c7463cdd7f7a1a3bc6268e5fd/PhpAmqpLib/Connection/AbstractConnection.php#L573-L609","documentation":"wait_frame() reads frames from the wire; if the connection's input reader is null the connection is already dead, so the library marks the connection disconnected and throws AMQPConnectionClosedException('Broken pipe or closed connection'). It signals the TCP connection to the broker is gone.","triggerScenarios":"Calling wait_channel()/wait_frame() after the broker closed the socket or the OS reported EPIPE; using a connection object after a previous I/O failure; network drop or broker restart mid-operation.","commonSituations":"Long-lived consumers whose connection died (idle timeout, firewall dropping TCP, RabbitMQ restart); publishing after the socket closed; Docker/NAT environments with aggressive idle timeouts.","solutions":["Catch AMQPConnectionClosedException and reconnect: recreate the connection, channels and consumers, then retry the operation","Enable heartbeats ($config->setHeartbeat(30)) and a keepalive so dead connections are detected and maintained","Check broker logs / network (idle timeouts, load balancer) for why the socket closed","Wrap long-running consume loops in a reconnect supervisor"],"exampleFix":"// before\n$channel->wait();\n// after\ntry {\n    $channel->wait();\n} catch (AMQPConnectionClosedException $e) {\n    $connection = reconnect(); // rebuild connection + channel\n    $channel->basic_consume(...); // re-register consumers\n}","handlingStrategy":"try-catch","validationCode":"if (!$connection->isConnected()) {\n    [$connection, $channel] = reconnect();\n}","typeGuard":null,"tryCatchPattern":"try {\n    $channel->wait();\n} catch (AMQPConnectionClosedException $e) {\n    [$connection, $channel] = reconnectWithBackoff();\n    $channel->basic_consume($queue, ..., $callback);\n}","preventionTips":["Enable heartbeats and TCP keepalive to detect/maintain dead connections","Run consume loops inside a supervisor that reconnects on closed-connection exceptions","Re-declare queues/consumers after reconnect","Check firewall/LB idle timeout settings vs heartbeat interval"],"tags":["php","network","connection"],"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"}