Validate received messages cannot be added to Batch

implemented

Author: SeanFeldman

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

@@ -4,6 +4,7 @@
 using Microsoft.Azure.Amqp;
 using Microsoft.Azure.Amqp.Framing;
 using Microsoft.Azure.ServiceBus.Amqp;
+using Microsoft.Azure.ServiceBus.Diagnostics;
 
 namespace Microsoft.Azure.ServiceBus.Core
 {
@@ -36,6 +37,8 @@ public bool TryAdd(Message message)
         {
             ThrowIfDisposed();
 
+            message.VerifyMessageIsNotPreviouslyReceived();
+
             var amqpMessage = AmqpMessageConverter.SBMessageToAmqpMessage(message);
 
             if (firstMessage == null)

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

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft. All rights reserved.
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
+using Microsoft.Azure.ServiceBus.Diagnostics;
+
 namespace Microsoft.Azure.ServiceBus.Core
 {
     using System;
@@ -16,6 +18,7 @@ namespace Microsoft.Azure.ServiceBus.Core
     using Microsoft.Azure.ServiceBus.Amqp;
     using Microsoft.Azure.ServiceBus.Primitives;
 
+
     /// <summary>
     /// The MessageSender can be used to send messages to Queues or Topics.
     /// </summary>
@@ -236,7 +239,7 @@ public async Task SendAsync(IList<Message> messageList)
         {
             this.ThrowIfClosed();
 
-            var count = MessageSender.ValidateMessages(messageList);
+            var count = MessageSender.VerifyMessagesAreNotPreviouslyReceived(messageList);
             MessagingEventSource.Log.MessageSendStart(this.ClientId, count);
 
             bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
@@ -275,7 +278,6 @@ public async Task SendAsync(Batch batch)
         {
             this.ThrowIfClosed();
 
-            //var count = MessageSender.ValidateMessages(messageList);
 //            MessagingEventSource.Log.MessageSendStart(this.ClientId, count);
 
 //            var isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
@@ -335,7 +337,7 @@ public async Task<long> ScheduleMessageAsync(Message message, DateTimeOffset sch
             }
 
             message.ScheduledEnqueueTimeUtc = scheduleEnqueueTimeUtc.UtcDateTime;
-            MessageSender.ValidateMessage(message);
+            message.VerifyMessageIsNotPreviouslyReceived();
             MessagingEventSource.Log.ScheduleMessageStart(this.ClientId, scheduleEnqueueTimeUtc);
             long result = 0;
 
@@ -484,7 +486,7 @@ protected override async Task OnClosingAsync()
             await this.RequestResponseLinkManager.CloseAsync().ConfigureAwait(false);
         }
 
-        static int ValidateMessages(IList<Message> messageList)
+        static int VerifyMessagesAreNotPreviouslyReceived(IList<Message> messageList)
         {
             var count = 0;
             if (messageList == null)
@@ -495,20 +497,12 @@ static int ValidateMessages(IList<Message> messageList)
             foreach (var message in messageList)
             {
                 count++;
-                ValidateMessage(message);
+                message.VerifyMessageIsNotPreviouslyReceived();
             }
 
             return count;
         }
 
-        static void ValidateMessage(Message message)
-        {
-            if (message.SystemProperties.IsLockTokenSet)
-            {
-                throw Fx.Exception.Argument(nameof(message), "Cannot send a message that was already received.");
-            }
-        }
-
         static void CloseSession(SendingAmqpLink link)
         {
             // Note we close the session (which includes the link).

src/Microsoft.Azure.ServiceBus/Extensions/MessageDiagnosticsExtensions.cs

@@ -6,6 +6,7 @@ namespace Microsoft.Azure.ServiceBus.Diagnostics
     using System;
     using System.Collections.Generic;
     using System.Diagnostics;
+    using Microsoft.Azure.ServiceBus.Primitives;
 
     public static class MessageExtensions
     {
@@ -15,20 +16,20 @@ public static class MessageExtensions
         /// <returns>New <see cref="Activity"/> with tracing context</returns>
         /// </summary>
         /// <remarks>
-        /// Tracing context is used to correlate telemetry between producer and consumer and 
+        /// Tracing context is used to correlate telemetry between producer and consumer and
         /// represented by 'Diagnostic-Id' and 'Correlation-Context' properties in <see cref="Message.UserProperties"/>.
-        /// 
+        ///
         /// .NET SDK automatically injects context when sending message to the ServiceBus (if diagnostics is enabled by tracing system).
-        /// 
+        ///
         /// <para>
         /// 'Diagnostic-Id' uniquely identifies operation that enqueued message
         /// </para>
         /// <para>
         /// 'Correlation-Context' is comma separated list of sting key value pairs represeting optional context for the operation.
         /// </para>
-        /// 
+        ///
         /// If there is no tracing context in the message, this method returns <see cref="Activity"/> without parent.
-        /// 
+        ///
         /// Returned <see cref="Activity"/> needs to be started before it can be used (see example below)
         /// </remarks>
         /// <example>
@@ -39,26 +40,26 @@ public static class MessageExtensions
         ///    var activity = message.ExtractActivity();
         ///    activity.Start();
         ///    Logger.LogInformation($"Message received, Id = {Activity.Current.Id}")
-        ///    try 
+        ///    try
         ///    {
         ///       // process message
         ///    }
         ///    catch (Exception ex)
         ///    {
         ///         Logger.LogError($"Exception {ex}, Id = {Activity.Current.Id}")
         ///    }
-        ///    finally 
+        ///    finally
         ///    {
         ///         activity.Stop();
         ///         // Activity is stopped, we no longer have it in Activity.Current, let's user activity now
         ///         Logger.LogInformation($"Message processed, Id = {activity.Id}, Duration = {activity.Duration}")
         ///    }
         /// }
         /// </code>
-        /// 
-        /// Note that every log is stamped with <see cref="Activity.Current"/>.Id, that could be used within 
+        ///
+        /// Note that every log is stamped with <see cref="Activity.Current"/>.Id, that could be used within
         /// any nested method call (sync or async) - <see cref="Activity.Current"/> is an ambient context that flows with async method calls.
-        /// 
+        ///
         /// </example>
 
         public static Activity ExtractActivity(this Message message, string activityName = null)
@@ -149,5 +150,14 @@ internal static bool TryExtractContext(this Message message, out IList<KeyValueP
             }
             return false;
         }
+
+
+        internal static void VerifyMessageIsNotPreviouslyReceived(this Message message)
+        {
+            if (message.SystemProperties.IsLockTokenSet)
+            {
+                throw Fx.Exception.Argument(nameof(message), "Cannot send a message that was already received.");
+            }
+        }
     }
 }

test/Microsoft.Azure.ServiceBus.UnitTests/Primitives/BatchTests.cs

@@ -29,5 +29,17 @@ public void Should_throw_if_batch_disposed()
                 Assert.Throws<Exception>(() => batch.TryAdd(new Message()));
             }
         }
+
+        [Fact]
+        public void Should_throw_when_trying_add_received_message_to_batch()
+        {
+            using (var batch = new Batch(100))
+            {
+                var message = new Message("test".GetBytes());
+                message.SystemProperties.LockTokenGuid = Guid.NewGuid();
+
+                Assert.Throws<ArgumentException>(() => batch.TryAdd(message));
+            }
+        }
     }
 }