Add Channel.ConsumeWithContext to be able to cancel delivering

[t2y]

This is a reference implementation for #124 (comment).

[lukebakken]

Thanks! We'll take a look as soon as we have time.

[Zerpet]

@lukebakken perhaps we could introduce this in a minor version, since it deprecates Consume() and adds ConsumeWithContext(). I'm also happy if we decide to ship this in 2.0, but in that case, we should consider cutting a v1.x branch before merging.

[Zerpet]

@t2y just dropping a message to let you know that I have not forgotten about this PR. During my QA of this PR, I found something odd regarding message re-queueing, and I'm investigating that before merging. Plus past weeks I was pulled onto something else.

May have found a problematic behaviour with this implementation

[t2y]

@Zerpet Thank you for testing! I'm going to fix it when the problem can be reproduced.

[Zerpet]

Never mind, the behaviour is correct. It is important to note that cancelling a consumer will not re-queue in-flight messages. The app using this library must close the amqp channel to re-queue in-flight messages.

I would like to have a check on context cancellation before calling ch.call(...), as discussed in #124, before merging this PR.

[Zerpet]

I realised that this PR changes the semantics of consume, into something that is not correct.

Previously, Consume would create a subcription in the server; such subscription will be valid until the channel is closed, or Cancel() was called. After this PR, the subcription will be valid until the channel is closed, or Cancel() is called, or the context gets cancelled.

We can't accept this PR with the new semantics, and it doesn't seem correct to use context cancellation to cancel a consumer. The context passed to ConsumeWithContext() should be used to cancel the call to ConsumeWithContext(), but it should not cancel a consumer after ConsumeWithContext() has returned successfully.

In other words, the go routine initiated in ConsumeWithContext() will have an impact on the consumer after ConsumeWithContext() returns successfully, and that is not correct. A consumer should be able to keep on receiving messages after the context in ConsumeWithContext is cancelled.

From the conversation in #124, the best way forward is to check for context cancellation before calling ch.call(...).

In addition, and optionally, we could wrap ch.call(...) in a go routine and observe context cancellation, or a signal from the wrapped call. The clean up tasks would be to close the AMQP channel. I'm not sure about this bit, please hold on from implementing this bit.

[t2y]

Thank you for describing the background. I will follow you if this PR is not valid for consumption. Before that, I want to confirm a little more as a reference.

I understood existing Consume() is different from ConsumeWithContext() . Of course, it's important not to change the semantics of consume, too.

At first, I thought Consume() is similar to the Generators pattern in Go. So, I'm considering that canceling with Context is useful for Go developers since they can see it's a general concurrency pattern. I guess many Go developers expect canceling Context stops the subscription. What do you think?

[batazor]

I agree, by canceling Context I hope that everything below will also be canceled

[lukebakken]

We can't accept this PR with the new semantics, and it doesn't seem correct to use context cancellation to cancel a consumer.

I would expect context cancellation to "cancel everything" ... i.e. do a Basic.Cancel operation.

[Zerpet]

At first, I thought Consume() is similar to the Generators pattern in Go.

I believe this is a coincidence. The original author of this package @streadway wrote it way before contexts made it to Go. Even if the generators pattern can be implemented w/o contexts, I don't think it was the original intention.

I agree, by canceling Context I hope that everything below will also be canceled

I would expect context cancellation to "cancel everything" ... i.e. do a Basic.Cancel operation.

The "catch" here IMO is that ConsumeWithContext has returned, and you get to work with the returned value. There's nothing "below". Compare this example with a Dial(context, address) function, e.g. with the dialer from the standard library. The context is used as a timeout for the dialer, but once dialer has returned, context cancellation does not affect the dialed function.

That being said, if folks find useful to cancel a consumer after the context expires, even when ConsumeWithContext has returned, I'd be happy to introduce a new function, without deprecating Consume, since both have different semantics.

[t2y]

That being said, if folks find useful to cancel a consumer after the context expires, even when ConsumeWithContext has returned, I'd be happy to introduce a new function, without deprecating Consume, since both have different semantics.

I agree with you. I updated the description of functions (I'd like to ask anyone to rewrite the proper content since I'm not a good English writer).

[Zerpet]

We haven't seen much activity on this PR for a while. Based on the previous feedback, I'm going to merge as-is and make some adjustments to docs and to original Consume function. We should be able to release this in a minor version of 1.x.

Thanks everyone for getting involved!