Receive calls could timeout when they are made concurrently and auto-…

…flow is off.
Azure · Nov 4, 2020 · ebb18dd · ebb18dd
1 parent 69fddda
commit ebb18dd
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 3 deletions.
diff --git a/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs
@@ -563,7 +563,9 @@ int GetOnDemandReceiveCredit()
                     currentCredit < MaxCreditForOnDemandReceive)
                 {
                     int needCredit = Math.Min(this.waiterList.Count, MaxCreditForOnDemandReceive) - currentCredit;
-                    if (this.waiterList.Count <= CreditBatchThreshold || needCredit % CreditBatchThreshold == 0)
+                    if (this.waiterList.Count <= CreditBatchThreshold ||
+                        currentCredit == 0 ||
+                        needCredit % CreditBatchThreshold == 0)
                     {
                         credit = currentCredit + needCredit;
                     }
@@ -574,7 +576,9 @@ int GetOnDemandReceiveCredit()
                 if (totalRequestedMessageCount > currentCredit)
                 {
                     int needCredit = totalRequestedMessageCount - currentCredit;
-                    if (this.waiterList.Count <= PendingReceiversThreshold || this.waiterList.Count % PendingReceiversThreshold == 0)
+                    if (this.waiterList.Count <= PendingReceiversThreshold ||
+                        currentCredit == 0 ||
+                        this.waiterList.Count % PendingReceiversThreshold == 0)
                     {
                         credit = currentCredit + needCredit;
                     }

diff --git a/Microsoft.Azure.Amqp/Properties/Version.cs b/Microsoft.Azure.Amqp/Properties/Version.cs
@@ -3,4 +3,4 @@
 using System.Runtime.InteropServices;
 
 [assembly: AssemblyVersion("2.4.0.0")]
-[assembly: AssemblyInformationalVersion("2.4.7")]
+[assembly: AssemblyInformationalVersion("2.4.8")]
diff --git a/test/TestCases/AmqpLinkTests.cs b/test/TestCases/AmqpLinkTests.cs
@@ -440,6 +440,58 @@ public void AmqpConcurrentConnectionsTest()
             Assert.True(lastException == null, string.Format("Failed. Last exception {0}", lastException == null ? string.Empty : lastException.ToString()));
         }
 
+        [Fact]
+        public void NonPrefetchConcurrentReceiveTest()
+        {
+            string queue = "NonPrefetchConcurrentReceiveTest";
+            broker.AddQueue(queue);
+
+            AmqpConnection connection = AmqpUtils.CreateConnection(addressUri, null, false, null, (int)AmqpConstants.DefaultMaxFrameSize);
+            connection.Open();
+
+            AmqpSession session = connection.CreateSession(new AmqpSessionSettings());
+            session.Open();
+
+            ReceivingAmqpLink rLink = new ReceivingAmqpLink(session, AmqpUtils.GetLinkSettings(false, queue, SettleMode.SettleOnSend, 0));
+            rLink.Open();
+
+            bool done = false;
+            int count = 0;
+            Task sendTask = Task.Run(() =>
+            {
+                SendingAmqpLink sLink = new SendingAmqpLink(session, AmqpUtils.GetLinkSettings(true, queue, SettleMode.SettleOnReceive));
+                sLink.Open();
+                while (!done)
+                {
+                    AmqpMessage message = AmqpMessage.Create(new AmqpValue() { Value = "Test" });
+                    Outcome outcome = sLink.EndSendMessage(sLink.BeginSendMessage(message, new ArraySegment<byte>(new byte[2]), new ArraySegment<byte>(), TimeSpan.FromSeconds(5), null, null));
+                    Assert.True(outcome.DescriptorCode == Accepted.Code, "message is not accepted.");
+                }
+            });
+
+            Task.WaitAll(Enumerable.Range(0, 30).Select(i =>
+            {
+                return Task.Run(async () =>
+                {
+                    while (!done)
+                    {
+                        var msg = await rLink.ReceiveMessageAsync(TimeSpan.FromSeconds(10));
+                        Assert.NotNull(msg);
+                        rLink.AcceptMessage(msg, false);
+                        if (Interlocked.Increment(ref count) > 10000)
+                        {
+                            break;
+                        }
+                    }
+                });
+            }).ToArray());
+
+            done = true;
+            sendTask.Wait();
+
+            connection.Close();
+        }
+
         [Fact]
         public void AmqpSequenceNumberWrapAroundTest()
         {