support manualAck attribute for in-proc

[ishank12]

As discussed in the issue #151

It is common scenario that message ack / reject would need to be done by the function code and not always by the host.
This PR is fixing that problem by introducing a new trigger attribute DisableAck.
If this attribute is set to true, host would not perform acknowledgements for the processed message and function code need to do it.
The default value of this attribute is false, so existing apps would be working as is.

Since the Ack/Reject needs to be done on the exact same channel instance which was created by the host, the host share this channel instance encapsulated in a class RabbitMQMessageActions and pass it as trigger binding.

below code is for reference how it can be used.

[FunctionName("Function1")]
public void Run([RabbitMQTrigger("test11", ConnectionStringSetting = "rabbitmqconnsetting", DisableAck = true)]string myQueueItem,
    string consumerTag,
    ulong deliveryTag,
    RabbitMQMessageActions messageActions,
    ILogger log)
{
    log.LogInformation($"C# Queue trigger function processed: {myQueueItem}, deliveryTag: {deliveryTag}");
    
    Console.WriteLine("Acknowleding...");
    messageActions.Acknowledge();
}

Note:
Now, since the host is passing the channel instance to the function code, this is currently only supporting in-proc and C# functions only.
for non C# functions DisableAck should be always false.

[vivekjilla]

Can you add detailed description of changes, testing done etc in PR description above, for future reference

[aloiva]

Wondering if we can reduce the function app code by adding deliveryTag in the messageActions itself as a private member since each execution has a new instance of RabbitMQMessageActions that pertains to the current message. Would make it much easier this way instead of mandating two parameters to make use of this feature:

[FunctionName("Function1")]
public void Run([RabbitMQTrigger("test11", ConnectionStringSetting = "rabbitmqconnsetting", ManualAck = true)]string myQueueItem,
    RabbitMQMessageActions messageActions,
    ILogger log)
{
    log.LogInformation($"C# Queue trigger function processed: {myQueueItem}, deliveryTag: {deliveryTag}, messageActions: ");
    
    Console.WriteLine("Acknowleding...");
    messageActions.BasicAck();
    // messageActions.BasicReject(requeue=true);
}```

[aloiva]

I see a complication in this implementation. If the customer code acks the message and fails at a later point, the extension tries to reject the already acknowledged message and requeue it as a retry mechanism. Would this throw an error since by definition the message would already be deleted after ack? It would be better to entirely avoid such complicated scenarios by maybe adding a mechanism that tracks if a message has been acked or not.

[ishank12]

Wondering if we can reduce the function app code by adding deliveryTag in the messageActions itself as a private member since each execution has a new instance of RabbitMQMessageActions that pertains to the current message. Would make it much easier this way instead of mandating two parameters to make use of this feature:

[FunctionName("Function1")]
public void Run([RabbitMQTrigger("test11", ConnectionStringSetting = "rabbitmqconnsetting", ManualAck = true)]string myQueueItem,
    RabbitMQMessageActions messageActions,
    ILogger log)
{
    log.LogInformation($"C# Queue trigger function processed: {myQueueItem}, deliveryTag: {deliveryTag}, messageActions: ");
    
    Console.WriteLine("Acknowleding...");
    messageActions.BasicAck();
    // messageActions.BasicReject(requeue=true);
}

@olalavi , each execution will not have new instance of RabbitMQMessageActions. The channel instance is only one throughout the lifecycle of extension and so is the RabbitMQMessageActions created only once.
it would be inefficient to create separate instance for each new message.

But it can be done

[ishank12]

I see a complication in this implementation. If the customer code acks the message and fails at a later point, the extension tries to reject the already acknowledged message and requeue it as a retry mechanism. Would this throw an error since by definition the message would already be deleted after ack? It would be better to entirely avoid such complicated scenarios by maybe adding a mechanism that tracks if a message has been tracked or not.

Hey @aloiva ,
Yes that is a good point,
if BasicAck or Reject happens on the same delivery tag twice in any sequence, it will throw exception on channel and channel will be closed.
I'll try to address this.

[aloiva]

Yes that is a good point,
if BasicAck or Reject happens on the same delivery tag twice in any sequence, it will throw exception on channel and channel will be closed.
I'll try to address this.

@ishank12 , another scenario to address would be where manualAck is set to false (or disableAck is set to true) and yet the customer makes use of MessageActions to ack or reject in their function code. Extension will also try to do the same after function execution and will throw an error. Instead of the methods in MessageActions being independent, making them dependent on manualAck/disableAck bool might fix this. Or making how MessageActions is received by the function code based on manualAck/disableAck (maybe return null in such cases instead of always returning the initialised object?) Just throwing out ideas here..

[ishank12]

/azp run rabbitmq-extension-linux.public

[azure-pipelines]

Commenter does not have sufficient privileges for PR 255 in repo Azure/azure-functions-rabbitmq-extension

[aloiva]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 2 pipeline(s).