Ack messages with multiple connections about aioamqp HOT 16 CLOSED

Hello @thijsterlouw,

Thank you for reporting your use case. This issue really shows that we didn't solve the problem correctly. For instance we should probably pass the channel instance to ack this specific message.
Problem is: this channel is currently overflooded by incomming message and won't send any message to the server :/

from aioamqp.

Hi @dzen , thanks for your quick reply. I've created a little bit of Python code to test the idea, and it seems wrapping my code in Objects would be a good way to go forward. That way I can pass my context around in the objects and have the object just create a reply channel to ack the messages (like #39 suggested)

import asyncio


def tester():
    print('tester')
    x = MyClass(1)
    y = MyClass(2)
    print('created classes')
    print(asyncio.iscoroutinefunction(x.f))
    val_x = yield from x.f()
    val_y = yield from y.f()
    print('x: {0}, y: {1}'.format(val_x, val_y))


class MyClass:
    def __init__(self, i):
        self.i = i

    @asyncio.coroutine
    def f(self):
        return 'hello world {0}'.format(self.i)


if __name__ == '__main__':
    print('main')
    loop = asyncio.get_event_loop()
    loop.run_until_complete(tester())
    loop.run_forever()
    loop.stop()

from aioamqp.

I have everything setup nicely. One instance of a class for each server, which wraps the callback and provides the necessary context. Acking now works great (even when using the same channel). So for me: problem solved :)

Initially I had a small application bug where I didn't ack invalid events, but that was easily fixed.

from aioamqp.

Hi @thijsterlouw,

I tried a little thing: the callback is converted as a Task, and the channel instance is passed to the callback. This way it's possible to ack the message from the coroutine called when a message is received. The only problem is that we will create a Task for each received message. The eventloop could be very BIG if your coroutine is slow.

I've got a publisher which publish at 20k msg/s. My computer goes very slow :(

from aioamqp.

from aioamqp.

@mpaolini hello. Any thought on this PR / Issue ?

from aioamqp.

@dzen will have a look later this weekend

from aioamqp.

Hello @thijsterlouw,

Did you tried this branch against your code ?

beware, the definition of your callback should be like:

@asyncio.coroutine
def callback(channel, body, envelope, properties):
    # stuff
    yield from channel.basic_client_ack(envelope.delivery_tag)

from aioamqp.

Hi
I tried out that branch and I have a question. How to perform BASIC_ACK, because basic_client_ack and basic_server_ack both did not work?

from aioamqp.

do you have any problem using basic_client_ack ? what's the error generated ?

from aioamqp.

Yes I have, it does not work. No error, no ack, task still in queue.

from aioamqp.

Can you paste some code ?

from aioamqp.

@dzen sorry for missleading you, I forgot about yield from and when use yield from ch.basic_client_ask... it is working now. sorry again.

from aioamqp.

No problem. For now I didn't wrote enough documentation. It's planned

from aioamqp.

hello @thijsterlouw . Did you try the latest release ? (0.5.1, since we got a packaging problem) ?

from aioamqp.

@thijsterlouw you now have the channel handle to ack a message correctly.

from aioamqp.