Fix regression introduced in Sendpath because of an earlier checkin (#74)

Author: vinaysurya

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

@@ -34,47 +34,43 @@ static class AmqpMessageConverter
         public static AmqpMessage BrokeredMessagesToAmqpMessage(IEnumerable<BrokeredMessage> brokeredMessages, bool batchable)
         {
             AmqpMessage amqpMessage = null;
+            AmqpMessage firstAmqpMessage = null;
             BrokeredMessage firstBrokeredMessage = null;
             List<Data> dataList = null;
             int messageCount = 0;
             foreach (var brokeredMessage in brokeredMessages)
             {
-                if (firstBrokeredMessage == null)
-                {
-                    firstBrokeredMessage = brokeredMessage;
-                }
+                messageCount++;
 
-                if (messageCount == 1)
+                amqpMessage = AmqpMessageConverter.ClientGetMessage(brokeredMessage);
+                if (firstAmqpMessage == null)
                 {
-                    dataList = new List<Data> { ToData(amqpMessage) };
+                    firstAmqpMessage = amqpMessage;
+                    firstBrokeredMessage = brokeredMessage;
+                    continue;
                 }
 
-                amqpMessage = AmqpMessageConverter.ClientGetMessage(brokeredMessage);
-
-                if (messageCount > 1)
+                if (dataList == null)
                 {
-                    // ReSharper disable once PossibleNullReferenceException
-                    dataList.Add(ToData(amqpMessage));
+                    dataList = new List<Data>() { ToData(firstAmqpMessage) };
                 }
 
-                messageCount++;
+                dataList.Add(ToData(amqpMessage));
             }
 
-            if (messageCount == 1)
+            if (messageCount == 1 && firstAmqpMessage != null)
             {
-                amqpMessage.Batchable = batchable;
-                return amqpMessage;
+                firstAmqpMessage.Batchable = batchable;
+                return firstAmqpMessage;
             }
 
             amqpMessage = AmqpMessage.Create(dataList);
             amqpMessage.MessageFormat = AmqpConstants.AmqpBatchedMessageFormat;
 
-            // ReSharper disable once PossibleNullReferenceException
             if (firstBrokeredMessage.MessageId != null)
             {
                 amqpMessage.Properties.MessageId = firstBrokeredMessage.MessageId;
             }
-
             if (firstBrokeredMessage.SessionId != null)
             {
                 amqpMessage.Properties.GroupId = firstBrokeredMessage.SessionId;

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

@@ -75,6 +75,7 @@ internal static async Task<IEnumerable<BrokeredMessage>> ReceiveMessagesAsync(Me
                 }
             }
 
+            VerifyUniqueMessages(messagesToReturn);
             Log($"Received {messagesToReturn.Count} messages");
             return messagesToReturn;
         }
@@ -100,6 +101,7 @@ internal static async Task<IEnumerable<BrokeredMessage>> PeekMessagesAsync(Messa
                 }
             }
 
+            VerifyUniqueMessages(peekedMessages);
             Log($"Peeked {peekedMessages.Count} messages");
             return peekedMessages;
         }
@@ -127,5 +129,20 @@ internal static async Task DeferMessagesAsync(MessageReceiver messageReceiver, I
             await messageReceiver.DeferAsync(messages.Select(message => message.LockToken));
             Log($"Deferred {messages.Count()} messages");
         }
+
+        static void VerifyUniqueMessages(List<BrokeredMessage> messages)
+        {
+            if (messages != null && messages.Count > 1)
+            {
+                HashSet<long> sequenceNumbers = new HashSet<long>();
+                foreach (var message in messages)
+                {
+                    if (!sequenceNumbers.Add(message.SequenceNumber))
+                    {
+                        throw new Exception($"Sequence Number '{message.SequenceNumber}' was repeated");
+                    }
+                }
+            }
+        }
     }
 }