Recovery from ChannelWrongStateError

[colingoodman]

I have a service that publishes messages to a RabbitMQ server and need to ensure a high level of resilience. In the event the RabbitMQ server needs to be restarted or there is a network issue, this client service needs to be able to automatically recover the connection. This seems like it should be simple, but attempting to implement some of the suggestions I have found have not yielded improvements.

Here is a simple example script:

def connect():
    parameters = pika.ConnectionParameters(...)
    connection = None
    while not connection:
        try:
            connection = pika.BlockingConnection(parameters)
        except Exception:
            print('Connection error')
    return connection

connection = connect()
while True:
    t.sleep(5)
    try:
        channel.basic_publish(
        exchange='test', routing_key=type, body=message)
        print(" [x] Sent %r:%r" % (type, message))
    except Exception as e:
        print(traceback.format_exc())
        connection = connect()

The connect function will connect forever until success. I have this set up so that if the while loop ever encounters an issue when publishing messages it will call this connect function. On startup this works fine, but I have found that if the connection is interrupted at any point after a successful connection, it will never be able to reconnect:

pika.exceptions.ChannelWrongStateError: Channel is closed.

It sounds like this is caused by the RabbitMQ server for some reason thinking that the connection is permanently ruined and rejecting it; I cannot figure out how to create a fresh connection without restarting my client entirely.

Thanks in advance for any assistance.

[lukebakken]

Thanks for using Pika and RabbitMQ.

Instead of copy-pasted code, please provide a git repository with code I can clone and run. As it is, your code will not run on its own and I don't have time to figure out how to make it work. I can probably guess but it's time taken away from reproducing and diagnosing your issue.

if the connection is interrupted at any point after a successful connection

Please let me know how you are interrupting the connection.

[colingoodman]

Thank you for the reply. Here is a slightly modified version of the code that will run. I have a docker container (docker run rabbitmq:3.11-management) for the server. This script is all I currently have; hopefully it properly expresses my question. It attempts to reconnect to the server if it encounters an issue publishing a message.

import pika
import time as t
import traceback

credentials = pika.PlainCredentials('username','password')  # change as necessary

def connect():
    parameters = pika.ConnectionParameters('localhost', 5672, '/', credentials, connection_attempts=5, retry_delay=1, heartbeat=0)
    connection = None
    while not connection:
        try:
            connection = pika.BlockingConnection(parameters)
        except Exception as e:
            print(f'Failed to connect.. {e}')
    return connection

connection = connect()
channel = connection.channel()
channel.exchange_declare(exchange='example_exchange', exchange_type='direct')

while True:
    t.sleep(1)
    try:
        channel.basic_publish(exchange='example_exchange', routing_key='a_routing_key', body='hello world')
        print('published a message')
    except Exception as e:
        print(traceback.format_exc())
        connection = connect()

Please let me know how you are interrupting the connection.

By restarting the RabbitMQ server, specifically stopping and starting the aforementioned docker container. If I do this the client script provided above will never be able to reconnect and will print the following message forever:

pika.exceptions.ChannelWrongStateError: Channel is closed.

How can I reconnect to the to the server without restarting this client script? I want the client to be able to automatically recover the connection.

[colingoodman]

I found a script in a Stack Overflow answer that is similar to the functionality in my script with a key difference being that it works. I made the following change to ensure it will attempt to reconnect until success:

    def connect(self):
        while not self._conn or self._conn.is_closed:
            try:
                t.sleep(0.5)
                self._conn = pika.BlockingConnection(self._params)
                self._channel = self._conn.channel()
                self._channel.exchange_declare(exchange=self.EXCHANGE,
                                           exchange_type=self.TYPE)
            except Exception as e:
                print(e)
                t.sleep(1)

There are a couple of discrepancies that I can tell:

• The stack overflow script also checks for self._conn.is_closed
• The stack overflow script declares a new exchange every time a new BlockConnection object is created, I was only doing it once