Isolate KeyErrors raised during frame dispatch

[foolswood]

Should dispatch_frame (which may delegate to user callbacks) raise a KeyError, it isn't an unknown channel, so the log is misleading, and the exception should not be swallowed.

[dzen]

Hello,

When/How did you get this error ?

[foolswood]

Had a bug in a consume callback that happened to raise a KeyError, which caused the "unknown channel" log to occur.

[dzen]

Would you please add a test with a callback that raise a KeyError ?

[foolswood]

I can, but without reworking the test system it will pass on both branches, given that it trades a log for an internal error (which becomes a log in the tests at the moment).

[dzen]

The problem is that we're in an async context when consuming messages from RabbitmMQ so as when receiving anything from it. Notifying the code of an error in the callback is hard.

Maybe the error_callback should be fired ? I'm really not sure this is a good solution..

Note: the task here ( https://github.com/Polyconseil/aioamqp/blob/master/aioamqp/protocol.py ) is not supervised. so if it stop we will be stuck. The exception catched here ( https://github.com/Polyconseil/aioamqp/blob/master/aioamqp/protocol.py#L261 ) should stop everything isn't it ?

[foolswood]

If the user callback fails to handle a message, that is bad, but I wouldn't have thought it was terminal for the whole server (or even necessarily subsequent consumes from the same queue).

[foolswood]

This makes it consistent with the behaviour for all errors that aren't KeyError. Deciding what the behaviour should be is a slightly bigger task.

[foolswood]

I am happy to follow up with a pull request addressing this issue once consensus is reached.

[dzen]

Hello @foolswood,

There are two ideas that came up from this PR:

• first of all: I think that, at a higher level, we should probably raise an Exception (something like ConsumptionStopped) when something went wrong in the callback.
• then, we should be able to create a test for keyerrors in the callback and apply your nice patch about channel verification in yield from self.channels[frame.channel].dispatch_frame(frame)

I've still two interrogations:

• About the async context, we probably should have one way to warn about the errors, but how ? do we than:
• an error callback
• shall basic_consume have to return a Future on which we will do a future.set_exception() if the callback fails ? I think it's not really a good way or a nice API :(
  • Do we need to call channel.basic_cancel to warn the server that we won't be able to unqueue message with this code ?

[dzen]

@foolswood I edited this comment twice. @mastak any idea ?

[mastak]

Hello,

As for me the idea of a Future as a result of basic_consume not quite good. What will happen to consumer_tag (basic_consume generate and return consumer_tag)? Also main mission of the basic_consume – set callback of new messages. And if basic_consume raise some exception, it must be part of basic_consume process: set callback (but not part of user callback).

So, in my opinion, error callback/exception handler would be better.

PS asyncio has exception_handler too :) https://docs.python.org/3/library/asyncio-eventloop.html#error-handling-api

[dzen]

Yes, but the exception will be raised in the callback and then in the basic_deliver process, not when registering the callback to the queue using basic_consume.

[dzen]

I'm still not convinced by the callback parameter.

One of the first issue I addressed was the consumtion API: https://github.com/Polyconseil/aioamqp/issues/2

In my opinion, callback are not really a good design, and asyncio is built to use futures, or task, to have an iterative program, and avoid the use of callbacks.

On its first versions, I tried to use an asyncio.Queue to address the issue. ( https://github.com/Polyconseil/aioamqp/blob/32d60abebd5c232595d4eb2827434cb6e10a2f16/aioamqp/channel.py#L359 )

What is the API we WANT to use in the client code ? we probably should start there.

Do we really want to have a callback, or an API like:

    class Message():
        #body, envelope, properties

    @asyncio.coroutine
    def unqueue(consumer):
        # A receive should be an iterable
        print(receiver.consumer_tag)
        for message in consumer:
            try:
                process(message)
            except WhatEver:
                yield from consumer.channel.basic_cancel()

    @asyncio.coroutine
    def initialize()
        transport, protocol = yield from aioamqp.connect()
        channel = yield from protocol.channel()

        yield from channel.queue_declare(queue_name='hello')

        consumer = yield from channel.basic_consume(queue_name='hello')

        asyncio.async(unqueue(consumer))

The problem of this kind of API is that we will increase the logic/complexity in the lib which will probably have to use asyncio synchronization methods to not read all messages from the current channel and wait until we unqueues this message.

[foolswood]

I fiddled with a similar scheme in one of my toy projects, it ended up being quite painful as asyncio is already using the generator mechanisms and you ended up with a generator that yielded coroutines.

It also increases the complexity in the lib user code, you end up with consumes looking (from their perspective) like get() only more complex to set up.

[foolswood]

Given that this doesn't change the more general error handling situation, could we fix this and come back to the wider point?

[dzen]

Yes we have to!

On Fri, Mar 4, 2016 at 10:21 AM, David Honour notifications@github.com wrote:

Given that this doesn't change the more general error handling situation, could we fix this and come back to the wider point?

— Reply to this email directly or view it on GitHub https://github.com/Polyconseil/aioamqp/pull/79#issuecomment-192201032.

Benoît CALVEZ Polyconseil | 26 rue de Berri | 75008 Paris

[dzen]

Merged ! with https://github.com/Polyconseil/aioamqp/commit/bae422c2e77fbfab43520b61ff0bb39decb3e2d9