AMQPConnectionError using Pika and RabbitMQ with Asyncore - why?

Question

Why am I getting an AMQPConnectionError with Asyncore but not with BlockingConnection?

If it's just 'Asyncore doesn't work in Windows' then so be it, though I've yet to find anything that forbids its use. (This issue is platform independent.) For ease of migration I'd like to use async libraries that are available on both Python 2.7 and Python 3.4, and Asyncore should work here.

I'm using RabbitMQ 3.2.4 with Python 2.7.6 and pika 0.9.13. User and admin run level didn't make a difference. Logger's presence or absence in the code has no bearing on the error, except the updated warning message above. Same error is seen in Linux (Ubuntu 14.04) and Windows 7, so it's not a platform problem.

Because the performance of pika is rather poor using BlockingConnection, I wanted to try the Asyncore adapter instead. Seems pretty straightforward for a testbed setup (I tried giving it credentials though that shouldn't matter and the callbacks are stubbed out if not given... it fails either way.):

With BlockingConnection per the tutorial - it works, but has low throughput:

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

With AsyncoreConnection - all variants of this that I've tried fail immediately:

connection = pika.AsyncoreConnection(pika.ConnectionParameters(host='localhost'))

The error:

WARNING:pika.connection:Could not connect, 0 attempts left
Traceback (most recent call last):
  File "C:\workspace\send.py", line 8, in <module>
    connection = pika.AsyncoreConnection(pika.ConnectionParameters(host='localhost'))
  File "C:\Python27\lib\site-packages\pika\adapters\asyncore_connection.py", line 135, in __init__
    stop_ioloop_on_close)
  File "C:\Python27\lib\site-packages\pika\adapters\base_connection.py", line 62, in __init__
    on_close_callback)
  File "C:\Python27\lib\site-packages\pika\connection.py", line 590, in __init__
    self.connect()
  File "C:\Python27\lib\site-packages\pika\connection.py", line 707, in connect
    self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 61, in wrapper
    return function(*tuple(args), **kwargs)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 92, in wrapper
    return function(*args, **kwargs)
  File "C:\Python27\lib\site-packages\pika\callback.py", line 232, in process
    callback(*args, **keywords)
  File "C:\Python27\lib\site-packages\pika\connection.py", line 1192, in _on_connection_error
    raise exceptions.AMQPConnectionError(self.params.connection_attempts)
pika.exceptions.AMQPConnectionError: 1

Answer 1

It actually looks like a bug in pika to me. Here's the connection.connect() code that's ultimately raising the exception:

def connect(self):
    """Invoke if trying to reconnect to a RabbitMQ server. Constructing the
    Connection object should connect on its own.

    """
    self._set_connection_state(self.CONNECTION_INIT)
    if self._adapter_connect():
        return self._on_connected()
    self.remaining_connection_attempts -= 1
    LOGGER.warning('Could not connect, %i attempts left',
                   self.remaining_connection_attempts)
    if self.remaining_connection_attempts:
        LOGGER.info('Retrying in %i seconds', self.params.retry_delay)
        self.add_timeout(self.params.retry_delay, self.connect)
    else:
        self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self)
        self.remaining_connection_attempts = self.params.connection_attempts
        self._set_connection_state(self.CONNECTION_CLOSED)

So, self._adapter_connect() is clearly not returning True, which suggests the connection is failing. Here's the AsyncoreConnection._adapter_connect code:

def _adapter_connect(self):
    """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting
    Pika's suggested buffer size for socket reading and writing. We pass
    the handle to self so that the AsyncoreDispatcher object can call back
    into our various state methods.

    """
    if super(AsyncoreConnection, self)._adapter_connect():
        self.socket = PikaDispatcher(self.socket, None, self._handle_events)
        self.ioloop = self.socket
        self._on_connected()

It doesn't return anything! So that if statement in connect is never going to be True. If I change the method to reflect the pattern all the other adapters use:

def _adapter_connect(self):
    """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting
    Pika's suggested buffer size for socket reading and writing. We pass
    the handle to self so that the AsyncoreDispatcher object can call back
    into our various state methods.

    """
    if super(AsyncoreConnection, self)._adapter_connect():
        self.socket = PikaDispatcher(self.socket, None, self._handle_events)
        self.ioloop = self.socket
        return True
    return False

It works fine. I would definitely file that bug!

Edit:

The bug appears to be fixed in the latest version (from github):

    def _adapter_connect(self):
        """Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting Pika's suggested buffer size for socket reading and writing. We pass the handle to self so that the AsyncoreDispatcher object can call back into our various state methods.

        """
        error = super(AsyncoreConnection, self)._adapter_connect()
        if not error:
            self.socket = PikaDispatcher(self.socket, None,
                                         self._handle_events)
            self.ioloop = self.socket
            self._on_connected()
        return error

Answer 2

try the steps mentioned below. I was facing the same issue on my centos machine.

1. sudo yum install rabbitmq-server
2. sudo service rabbitmq-server restart

Answer 3

Read this post : No handlers could be found for logger "pika.adapters.blocking_connection"

Fixed by adding:

import logging
logging.basicConfig()

EDIT

The issue has been reported https://github.com/pika/pika/issues/468

Answer 4

you have to start rabbitmq-server by running below command, then it will works

sudo systemctl start rabbitmq-server