Keep delivery count as-is when peeking a message (#590)

Fixes #589

* Verification test
* Differentiate between peek and receive operations
* Indicate peek operation
* Additional test case to verify behavior when message is received
* Test delivery count on peek against real queue
* Test delivery count on peeklock against real queue

Author: SeanFeldman

src/Microsoft.Azure.ServiceBus/Amqp/AmqpMessageConverter.cs

@@ -164,7 +164,7 @@ public static AmqpMessage SBMessageToAmqpMessage(SBMessage sbMessage)
             return amqpMessage;
         }
 
-        public static SBMessage AmqpMessageToSBMessage(AmqpMessage amqpMessage)
+        public static SBMessage AmqpMessageToSBMessage(AmqpMessage amqpMessage, bool isPeeked = false)
         {
             if (amqpMessage == null)
             {
@@ -229,7 +229,7 @@ public static SBMessage AmqpMessageToSBMessage(AmqpMessage amqpMessage)
 
                 if (amqpMessage.Header.DeliveryCount != null)
                 {
-                    sbMessage.SystemProperties.DeliveryCount = (int)(amqpMessage.Header.DeliveryCount.Value + 1);
+                    sbMessage.SystemProperties.DeliveryCount = isPeeked ? (int)(amqpMessage.Header.DeliveryCount.Value) : (int)(amqpMessage.Header.DeliveryCount.Value + 1);
                 }
             }
 

src/Microsoft.Azure.ServiceBus/Core/MessageReceiver.cs

@@ -1102,7 +1102,7 @@ protected virtual async Task<IList<Message>> OnPeekAsync(long fromSequenceNumber
                     {
                         var payload = (ArraySegment<byte>)entry[ManagementConstants.Properties.Message];
                         var amqpMessage = AmqpMessage.CreateAmqpStreamMessage(new BufferListStream(new[] { payload }), true);
-                        message = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage);
+                        message = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage, true);
                         messages.Add(message);
                     }
 

test/Microsoft.Azure.ServiceBus.UnitTests/AmqpConverterTests.cs

@@ -131,5 +131,39 @@ void SB_message_with_no_TTL_results_in_empty_Ampq_TTL()
 
             Assert.Null(amqpMessage.Header.Ttl);
         }
+
+        [Fact]
+        [DisplayTestMethodName]
+        void When_message_is_peeked_should_have_delivery_count_set_to_zero()
+        {
+            var messageBody = Encoding.UTF8.GetBytes("message1");
+
+            var amqpValue = new AmqpValue();
+            amqpValue.Value = new ArraySegment<byte>(messageBody);
+            var amqpMessage = AmqpMessage.Create(amqpValue);
+            amqpMessage.Header.DeliveryCount = 2;
+
+            var sbMessage = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage, isPeeked: true);
+            sbMessage.SystemProperties.SequenceNumber = 1L;
+
+            Assert.Equal(2, sbMessage.SystemProperties.DeliveryCount);
+        }
+
+        [Fact]
+        [DisplayTestMethodName]
+        void When_message_is_received_should_have_delivery_count_increased()
+        {
+            var messageBody = Encoding.UTF8.GetBytes("message1");
+
+            var amqpValue = new AmqpValue();
+            amqpValue.Value = new ArraySegment<byte>(messageBody);
+            var amqpMessage = AmqpMessage.Create(amqpValue);
+            amqpMessage.Header.DeliveryCount = 2;
+
+            var sbMessage = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage, isPeeked: false);
+            sbMessage.SystemProperties.SequenceNumber = 1L;
+
+            Assert.Equal(3, sbMessage.SystemProperties.DeliveryCount);
+        }
     }
 }

test/Microsoft.Azure.ServiceBus.UnitTests/QueueClientTests.cs

@@ -4,6 +4,7 @@
 namespace Microsoft.Azure.ServiceBus.UnitTests
 {
     using System.Collections.Generic;
+    using System.Linq;
     using System.Threading.Tasks;
     using Xunit;
 
@@ -31,6 +32,52 @@ async Task PeekLockTest(string queueName, int messageCount = 10)
             }
         }
 
+        [Theory]
+        [MemberData(nameof(TestPermutations))]
+        [DisplayTestMethodName]
+        async Task PeekDeliveryCountTest(string queueName)
+        {
+            var queueClient = new QueueClient(TestUtility.NamespaceConnectionString, queueName);
+            try
+            {
+                await TestUtility.SendMessagesAsync(queueClient.InnerSender, 1);
+
+                var message = await TestUtility.PeekMessageAsync(queueClient.InnerReceiver);
+
+                Assert.Equal(0, message.SystemProperties.DeliveryCount);
+            }
+            finally
+            {
+                var messageToDelete = await TestUtility.ReceiveMessagesAsync(queueClient.InnerReceiver, 1);
+                await TestUtility.CompleteMessagesAsync(queueClient.InnerReceiver, messageToDelete);
+
+                await queueClient.CloseAsync();
+            }
+        }
+
+        [Theory]
+        [MemberData(nameof(TestPermutations))]
+        [DisplayTestMethodName]
+        async Task PeekLockDeliveryCountTest(string queueName)
+        {
+            var queueClient = new QueueClient(TestUtility.NamespaceConnectionString, queueName);
+            try
+            {
+                await TestUtility.SendMessagesAsync(queueClient.InnerSender, 1);
+
+                var messages = await TestUtility.ReceiveMessagesAsync(queueClient.InnerReceiver, 1);
+
+                await TestUtility.CompleteMessagesAsync(queueClient.InnerReceiver, messages);
+
+                Assert.Equal(1, messages.First().SystemProperties.DeliveryCount);
+            }
+            finally
+            {
+                await queueClient.CloseAsync();
+            }
+        }
+
+
         [Theory]
         [MemberData(nameof(TestPermutations))]
         [DisplayTestMethodName]