Feature/can mark processed message complete when exception occurred (#16)

* added support for marking message as proccessed when exception occurred

* Added another test for success or fail as batch

* added comment

* Updated readme file on how to use MarkAsProccess

* updated comment and correct readme format

* Update src/QueueBatch/Impl/Listener.cs

Co-Authored-By: jeremy001181 <[email protected]>

* Update src/QueueBatch.Tests/Helpers.cs

Co-Authored-By: jeremy001181 <[email protected]>

* refactor according the review

Author: jeremy001181

README.md

@@ -35,10 +35,10 @@ public static void MyFunc([QueueBatchTrigger("myqueue")] IMessageBatch batch)
 }
 ```
 
-You can also acknowledge only some of the messages. The rest, will be retried in a similar manner to the regural `[QueueTrigger]`
+With `SuccessOrFailAsBatch` set to `false`, you can also acknowledge only some of the messages. The rest, will be retried in a similar manner to the regural `[QueueTrigger]`
 
 ```c#
-public static void MyFunc([QueueBatchTrigger("myqueue")] IMessageBatch batch)
+public static void MyFunc([QueueBatchTrigger("myqueue", SuccessOrFailAsBatch = false)] IMessageBatch batch)
 {
   foreach (var msg in batch.Messages)
   {

src/QueueBatch.Tests/Helpers.cs

@@ -39,11 +39,10 @@ public static async Task<List<CloudQueueMessage>> Drain(this CloudQueue queue, i
 
             return received;
         }
-
         public static async Task AssertIsEmpty(this CloudQueue queue)
         {
             var messages = await queue.GetMessagesAsync(32);
             CollectionAssert.IsEmpty(messages);
         }
     }
-}
+}

src/QueueBatch.Tests/IntegrationTests.cs

@@ -1,16 +1,82 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Text;
 using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs;
 using Microsoft.Azure.WebJobs.Host;
 using Microsoft.WindowsAzure.Storage.Queue;
+using Newtonsoft.Json;
 using NUnit.Framework;
 
 namespace QueueBatch.Tests
 {
     public class IntegrationTests : BaseTest
     {
+        [Test]
+        public async Task Messages_failed_as_batch_when_SuccessOrFailAsBatch_set_true()
+        {
+            const int count = 4;
+            await SendUnique(count);
+            await Batch.AddMessageAsync(new CloudQueueMessage("bad-guid"));
+
+            await RunHost<SuccessOrFailAsBatchTrue>(async () =>
+            {
+                await Poison.Drain(5);
+                await Batch.AssertIsEmpty();
+            });
+        }
+
+        [Test]
+        public async Task Can_mark_message_as_processed_while_other_cause_exception_when_SuccessOrFailAsBatch_set_false()
+        {
+            const int count = 4;
+            await SendUnique(count);
+            await Batch.AddMessageAsync(new CloudQueueMessage("bad-guid"));
+
+            await RunHost<SuccessOrFailAsBatchFalse>(async () =>
+            {
+                await Output.Drain(4);
+                await Poison.Drain(1);
+                await Batch.AssertIsEmpty();
+            });
+        }
+
+        public class SuccessOrFailAsBatchTrue
+        {
+            public static async Task Do(
+                  [QueueBatchTrigger(InputQueue, MaxBackOffInSeconds = 1, SuccessOrFailAsBatch = true)] IMessageBatch batch
+                , [Queue(OutputQueue)] CloudQueue output)
+            {
+                await SomeMessageInBatchCauseException(batch, output);
+            }
+        }
+
+        public class SuccessOrFailAsBatchFalse
+        {
+            public static async Task Do(
+                  [QueueBatchTrigger(InputQueue, MaxBackOffInSeconds = 1, SuccessOrFailAsBatch = false)] IMessageBatch batch
+                , [Queue(OutputQueue)] CloudQueue output)
+            {
+                await SomeMessageInBatchCauseException(batch, output);
+            }            
+        }
+
+        private static async Task SomeMessageInBatchCauseException(IMessageBatch batch, CloudQueue output)
+        {
+            var messages = batch.Messages.ToList();
+
+            foreach (var m in messages)
+            {
+                var content = Encoding.UTF8.GetString(m.Payload.Span);
+                var guid = Guid.Parse(content);
+                await output.AddMessageAsync(new CloudQueueMessage(m.Id));
+
+                batch.MarkAsProcessed(m);
+            }
+        }
+
         [Test]
         public async Task Simple_batch_dispatch()
         {
@@ -105,5 +171,6 @@ public static Task Do([QueueBatchTrigger(InputQueue, UseFasterQueues = true)] IM
                 return Task.CompletedTask;
             }
         }
+
     }
 }

src/QueueBatch.Tests/approved_api.txt

@@ -31,6 +31,7 @@ namespace QueueBatch
         public int ParallelGets { get; set; }
         public string QueueName { get; }
         public bool RunWithEmptyBatch { get; set; }
+        public bool SuccessOrFailAsBatch { get; set; }
         public bool UseFasterQueues { get; set; }
     }
 }

src/QueueBatch/Impl/BindingProvider.cs

@@ -54,7 +54,7 @@ await Task.WhenAll(
                 ? new QueueFunctionLogic(SdkQueue.CreateFast(messageQueue, cache), SdkQueue.CreateFast(poisonQueue, cache))
                 : new QueueFunctionLogic(new SdkQueue(messageQueue), new SdkQueue(poisonQueue));
 
-            return new TriggerBinding(context.Parameter, queue, TimeSpan.FromSeconds(attr.MaxBackOffInSeconds), attr.ParallelGets, attr.RunWithEmptyBatch, loggerFactory);
+            return new TriggerBinding(context.Parameter, queue, TimeSpan.FromSeconds(attr.MaxBackOffInSeconds), attr.ParallelGets, attr.SuccessOrFailAsBatch, attr.RunWithEmptyBatch, loggerFactory);
         }
         CloudQueue CreatePoisonQueue(CloudQueueClient queueClient, string name)
         {

src/QueueBatch/Impl/Listener.cs

@@ -17,6 +17,7 @@ class Listener : IListener
         readonly int maxRetries;
         readonly QueueFunctionLogic queue;
         readonly TimeSpan visibilityTimeout;
+        readonly bool successOrFailAsBatch;
         readonly bool shouldRunOnEmptyBatch;
         readonly ILoggerFactory loggerFactory;
         readonly Task<IRetrievedMessages>[] gets;
@@ -26,13 +27,14 @@ class Listener : IListener
         static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(10.0);
 
         public Listener(ITriggeredFunctionExecutor executor, QueueFunctionLogic queue,
-            TimeSpan maxBackOff, int maxRetries, TimeSpan visibilityTimeout, int parallelGets,
+            TimeSpan maxBackOff, int maxRetries, TimeSpan visibilityTimeout, int parallelGets, bool successOrFailAsBatch,
             bool shouldRunOnEmptyBatch, ILoggerFactory loggerFactory)
         {
             this.executor = executor;
             this.queue = queue;
             this.maxRetries = maxRetries;
             this.visibilityTimeout = visibilityTimeout;
+            this.successOrFailAsBatch = successOrFailAsBatch;
             this.shouldRunOnEmptyBatch = shouldRunOnEmptyBatch;
             this.loggerFactory = loggerFactory;
             gets = new Task<IRetrievedMessages>[parallelGets];
@@ -94,7 +96,7 @@ async Task Process()
 
                             try
                             {
-                                if (result.Succeeded)
+                                if (result.Succeeded || !successOrFailAsBatch)
                                 {
                                     await batch.Complete(CancellationToken.None);
                                 }
@@ -131,4 +133,4 @@ async Task Process()
 
         Task Delay(bool executionSucceeded, CancellationToken ct) => Task.Delay(backOff.GetNextDelay(executionSucceeded), ct);
     }
-}
+}

src/QueueBatch/Impl/TriggerBinding.cs

@@ -19,13 +19,15 @@ class TriggerBinding : ITriggerBinding
         readonly ILoggerFactory loggerFactory;
         readonly ParameterInfo param;
         readonly QueueFunctionLogic queue;
+        private bool successOrFailAsBatch;
 
-        public TriggerBinding(ParameterInfo param, QueueFunctionLogic queue, TimeSpan maxBackOff, int parallelGets, bool shouldRunOnEmptyBatch, ILoggerFactory loggerFactory)
+        public TriggerBinding(ParameterInfo param, QueueFunctionLogic queue, TimeSpan maxBackOff, int parallelGets, bool successOrFailAsBatch, bool shouldRunOnEmptyBatch, ILoggerFactory loggerFactory)
         {
             this.param = param;
             this.queue = queue;
             this.maxBackOff = maxBackOff;
             this.parallelGets = parallelGets;
+            this.successOrFailAsBatch = successOrFailAsBatch;
             this.shouldRunOnEmptyBatch = shouldRunOnEmptyBatch;
             this.loggerFactory = loggerFactory;
             BindingDataContract = new Dictionary<string, Type>(StringComparer.OrdinalIgnoreCase)
@@ -47,7 +49,7 @@ public Task<ITriggerData> BindAsync(object value, ValueBindingContext context)
 
         public Task<IListener> CreateListenerAsync(ListenerFactoryContext context)
         {
-            return Task.FromResult<IListener>(new Listener(context.Executor, queue, maxBackOff, 5, TimeSpan.FromSeconds(1), parallelGets, shouldRunOnEmptyBatch, loggerFactory));
+            return Task.FromResult<IListener>(new Listener(context.Executor, queue, maxBackOff, 5, TimeSpan.FromSeconds(1), parallelGets, successOrFailAsBatch, shouldRunOnEmptyBatch, loggerFactory));
         }
 
         public ParameterDescriptor ToParameterDescriptor()

src/QueueBatch/QueueBatchTriggerAttribute.cs

@@ -52,5 +52,10 @@ public QueueBatchTriggerAttribute(string queueName)
         /// Gets or sets the app setting name that contains the Azure Storage connection string.
         /// </summary>
         public string Connection { get; set; }
+
+        /// <summary>
+        /// If set to false, each message in same batch can be marked as processed independently and they will not affected when other messages cause exception, default to true
+        /// </summary>
+        public bool SuccessOrFailAsBatch { get; set; } = true;
     }
 }