What causes "Broken pipe or closed connection" in php-amqplib/php-amqplib?

Catch AMQPConnectionClosedException and reconnect: recreate the connection, channels and consumers, then retry the operation

php-amqplib/php-amqplib · critical · AMQPConnectionClosedException

Broken pipe or closed connection

Error message

Broken pipe or closed connection

What it means

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.

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

Example fix

// before
$channel->wait();
// after
try {
    $channel->wait();
} catch (AMQPConnectionClosedException $e) {
    $connection = reconnect(); // rebuild connection + channel
    $channel->basic_consume(...); // re-register consumers
}

Defensive patterns

Strategy: try-catch

Validate before calling

if (!$connection->isConnected()) {
    [$connection, $channel] = reconnect();
}

Try / catch

try {
    $channel->wait();
} catch (AMQPConnectionClosedException $e) {
    [$connection, $channel] = reconnectWithBackoff();
    $channel->basic_consume($queue, ..., $callback);
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Connection failures: ECONNREFUSED, ECONNRESET, and friends — why connections get refused, reset, or dropped.

Related errors

AI-assisted analysis of php-amqplib/php-amqplib@381b6f7c60 (2026-09-20). Data as JSON: /api/errors/26256d0d1ac30c1e. Report an issue: GitHub.

Appendix: source

Thrown at PhpAmqpLib/Connection/AbstractConnection.php:591

        $this->debug->debug_method_signature1($method_sig);

        return $pkt;
    }

    /**
     * Waits for a frame from the server
     *
     * @param int|float|null $timeout
     * @return Frame
     * @throws \Exception
     * @throws \PhpAmqpLib\Exception\AMQPTimeoutException
     * @throws AMQPRuntimeException
     */
    protected function wait_frame($timeout = 0): Frame
    {
        if (null === $this->input) {
            $this->setIsConnected(false);
            throw new AMQPConnectionClosedException('Broken pipe or closed connection');
        }

        $currentTimeout = $this->input->getTimeout();
        $this->input->setTimeout($timeout);

        try {
            $header = $this->input->readFrameHeader();
            $frame_type = $header['type'];
            if (!$this->constants->isFrameType($frame_type)) {
                throw new AMQPInvalidFrameException('Invalid frame type ' . $frame_type);
            }
            $size = $header['size'];

            // payload + ch
            $result = unpack('a' . $size . 'payload/Cch', $this->input->read(AMQPReader::OCTET + $size));
            $ch = $result['ch'];
            $frame = new Frame($frame_type, $header['channel'], $size, $result['payload']);
        } catch (AMQPTimeoutException $e) {

View on GitHub (pinned to 381b6f7c60)