Broken pipe or closed connection?

[tuongtd]

My consumer service get "Broken pipe or closed connection" error but I don't know why.
I follow this tutorial: https://www.rabbitmq.com/tutorials/tutorial-one-php.html

Everything seems to be running fine. But every time I try to push about 30k messages, the consumer console has stopped and throw error "Broken pipe or closed connection".

How to fix this problem? I tried use basic_qos(0, 200, false) but nothing changed. Do I have to change PHP configuration?
Thank!

PHP: 8.1
php-amqplib: 3.5.1
RabbitMQ: 3.10.6

[sabrimarzMed]

still facing same issue?

[tomaszgera]

Check your configuration. Version 3.5.4 got bug in AMQPConnectionConfig.php. Method setIsSecure() which is decide to use SSLConnection not change parameter networkProtocol:

public function setIsSecure(bool $isSecure): void
    {
        $this->isSecure = $isSecure;
    }

When it's passed AMQPConnectionConfig class to AMQPConnectionFactory.php networkProtocol parameter is still default 'tcp' and it's directly passed to AMQPSSLConnection.php by $config->getNetworkProtocol() :

if ($config->getIoType() === AMQPConnectionConfig::IO_TYPE_STREAM) {
            if ($config->isSecure()) {
                $connection = new AMQPSSLConnection(
                    $config->getHost(),
                    $config->getPort(),
                    $config->getUser(),
                    $config->getPassword(),
                    $config->getVhost(),
                    self::getSslOptions($config),
                    [
                        'insist' => $config->isInsist(),
                        'login_method' => $config->getLoginMethod(),
                        'login_response' => $config->getLoginResponse(),
                        'locale' => $config->getLocale(),
                        'connection_timeout' => $config->getConnectionTimeout(),
                        'read_write_timeout' => self::getReadWriteTimeout($config),
                        'keepalive' => $config->isKeepalive(),
                        'heartbeat' => $config->getHeartbeat(),
                    ],
                    $config->getNetworkProtocol(),
                    $config
                );
            }

You can fix it using method $config->setNetworkProtocol('ssl') on your config class

[tuongtd]

I run rabbitmq service and connect from PHP at localhost

$connection = new AMQPStreamConnection( '127.0.0.1', 5672, 'username', 'password', keepalive: true );

[tuongtd]

My cloud system sometimes seems to be full of memory and CPU, does that affect anything?

[sabrimarzMed]

I was facing this issue (in my producers) i got random broken pipe , missed HeartBeat exceptions

I tried couple workarounds like:
if ($rabbitMQConnection->isConnected()) {
rabbitMQConnection->reconnect()
}

Won't either work because of a possible Bug in the SDK or possibly due network issue that i couldn't diagnose.
I noticed that $rabbitMQConnection->isConnected() returns true while it's not probably true and and an Exception will be thrown. Basically, the idea is first to turn on HeartBeat (HeartBeat off isn't recommanded and you risk to have lot of zombie connections),and use checkHeartBeat() and then wherever you have Broken pipe you have to reconnect manually.

$connection = new AMQPStreamConnection( '127.0.0.1', 5672, 'username', 'password', heartbeat: true );

here is my full SYMFONY code that works that way ,hope it will inspire you somehow 😁😁 , attempt to use the connection, if exception is thrown $connection->reconnect() in catch clause ,and then with goto: it tries again

public function add(PushModal $pushModal): void
{
    $failed_once = false;

    /** @var AMQPStreamConnection $connection */
    $connection = $this->connectionFactory->createConnection();

    publish_attempt:
    try {
        $connection->checkHeartBeat();

        $channel = $connection->channel();

        $message = $pushModal->toArray();
        $msg = new AMQPMessage(json_encode($message, JSON_THROW_ON_ERROR));
        $channel->basic_publish($msg, 'notifications-exchange', 'notifications.push');

        /** @psalm-suppress InternalProperty */
        while (count($channel->callbacks)) {
            $channel->wait(null, false, 2);
        }

        $channel->close();
        unset($channel , $message);
    } catch (AMQPHeartbeatMissedException|AMQPConnectionClosedException|AMQPSocketException|AMQPTimeoutException|AMQPIOException) {
        $connection->reconnect();

        if (!$failed_once) {
            $failed_once = true;
            goto publish_attempt;
        }

        return;
    }
}