celery/py-amqp — pull request #48 by lezeroq (2014-08-08 12:53:44+00:00)
Title: Fix "amqp.exceptions.UnexpectedFrame: Received 0x69 while expecting 0xce" while using drain_events with timeout

Fix "amqp.exceptions.UnexpectedFrame: Received 0x69 while expecting 0xce" while using drain_events with timeout

This error is easy to reproduce on a slow connection to rabbitmq with many consumers and high message rate. If socket.timeout error occurs while reading payload:

def read_frame(self, unpack=unpack):
read = self._read
try:
frame_type, channel, size = unpack('>BHI', read(7, True))
payload = read(size) # If socket.timeout occurs here
ch = ord(read(1))
except socket.timeout:
raise

part of payload copied to the self._read_buffer, and next "read_frame" call tries to read frame_header (frame_type, channel, size) from self._read_buffer which contains part of payload from previous read_frame call.

Possible code sample for error reproducing:
while True:
try:
channel.connection.drain_events(timeout=0.1)
except socket.timeout:
# Do something else
# Do something else

Possible exceptions:
amqp.exceptions.UnexpectedFrame: Received 0x64 while expecting 0xce
amqp.exceptions.UnexpectedFrame: Received frame 1 while expecting type: 3