Closed connections not detected while waiting to consume messages

[tkram01]

When waiting on a message in a queue and RabbitMQ closes the connection because of missing heartbeats the closed connection is not detected and the program hangs.

Steps to reproduce:

1. Run the below code in a console
2. Ctrl-Z to stop the program
3. Watch the rabbitmq log for when it closes the connection
4. Resume the program (fg)
5. Insert something into the 'test' queue. Nothing will happen.

#!/usr/bin/env python
import logging
import rabbitpy

URL = 'amqp://guest:guest@localhost:5672/%2f?heartbeat=5'

logging.basicConfig(level=logging.INFO)

with rabbitpy.Connection(URL) as conn:
    with conn.channel() as channel:
        queue = rabbitpy.Queue(channel, 'test')
        queue.declare()
        # Exit on CTRL-C
        try:
            for message in rabbitpy.Queue(channel, 'test'):
                message.pprint(True)
                message.ack()
        except KeyboardInterrupt:
            logging.info('Exited consumer')

RabbitMQ output

=ERROR REPORT==== 31-Aug-2017::10:57:09 === closing AMQP connection <0.1120.0> (127.0.0.1:61721 -> 127.0.0.1:5672): missed heartbeats from client, timeout: 5s

[gmr]

Ok, so you're breaking the bidirectional RPC and expecting the client to know that the socket is closed when the program is not running. The program is not having an error raised on the socket and can't get the RPC from RabbitMQ to tell it to close the connection. I guess what would be useful in this scenario is to know why the IOLoop isn't seeing an error on the socket when it's trying to write to it.

Sure seems like an edge case to me, but I'll try to figure out what's going on with the socket. What OS are you doing this on? Will be helpful for reproducing the behavior...

[tkram01]

OS X 10.10

[nvllsvm]

Seems like a macOS-specific bug. I noticed a similar hanging behavior when publishing with confirmations.

https://gist.github.com/nvllsvm/65e5f78347fa58f7e48dcafa0bef1c0c

[nvllsvm]

Also encountered hanging when attempting to declare an exchange on a broken connection.

import os

import rabbitpy

conn = rabbitpy.Connection(os.environ['AMQP_URL'])
channel = conn.channel()

exchange = rabbitpy.Exchange(channel, 'amq.direct')
input('press enter')
exchange.declare(passive=True)

[gmr]

Hmm seems that Python 3.7 at least on MacOS is saying it has an implementation of Poll, which is weird, as that's a linux thing. I've changed the ordering of the ioloop poller selection and it's working for me on MacOS now. Will release a new version with this change.

[gmr]

This should be addressed in 2.0.1, released today.