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

…flow is off.
Azure · Nov 4, 2020 · c96c7cf · c96c7cf
1 parent de7083e
commit c96c7cf
Show file tree

Hide file tree

Showing 2 changed files with 65 additions and 6 deletions.
diff --git a/src/ReceivingAmqpLink.cs b/src/ReceivingAmqpLink.cs
@@ -391,7 +391,7 @@ public Outcome EndDisposeMessage(IAsyncResult result)
         public void AcceptMessage(AmqpMessage message)
         {
             bool settled = this.Settings.SettleType != SettleMode.SettleOnDispose;
-            this.AcceptMessage(message, settled, true);
+            this.AcceptMessage(message, settled, message.Batchable);
         }
 
         /// <summary>
@@ -415,7 +415,7 @@ public void RejectMessage(AmqpMessage message, Exception exception)
             Rejected rejected = new Rejected();
             rejected.Error = Error.FromException(exception);
 
-            this.DisposeMessage(message, rejected, true, false);
+            this.DisposeMessage(message, rejected, true, message.Batchable);
         }
 
         /// <summary>
@@ -424,7 +424,7 @@ public void RejectMessage(AmqpMessage message, Exception exception)
         /// <param name="message">The message to release.</param>
         public void ReleaseMessage(AmqpMessage message)
         {
-            this.DisposeMessage(message, AmqpConstants.ReleasedOutcome, true, false);
+            this.DisposeMessage(message, AmqpConstants.ReleasedOutcome, true, message.Batchable);
         }
 
         /// <summary>
@@ -441,7 +441,7 @@ public void ModifyMessage(AmqpMessage message, bool deliveryFailed, bool deliver
             modified.UndeliverableHere = deliverElseWhere;
             modified.MessageAnnotations = messageAttributes;
 
-            this.DisposeMessage(message, modified, true, false);
+            this.DisposeMessage(message, modified, true, message.Batchable);
         }
 
         /// <summary>
@@ -747,7 +747,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;
                     }
@@ -758,7 +760,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/test/TestCases/AmqpLinkTests.cs b/test/TestCases/AmqpLinkTests.cs
@@ -437,6 +437,61 @@ public async Task 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(async () =>
+            {
+                SendingAmqpLink sLink = new SendingAmqpLink(session, AmqpUtils.GetLinkSettings(true, queue, SettleMode.SettleOnReceive));
+                sLink.Open();
+                while (!done)
+                {
+                    await Task.WhenAll(Enumerable.Range(0, 6).Select(async i =>
+                    {
+                        AmqpMessage message = AmqpMessage.Create(new AmqpValue() { Value = "Test" });
+                        Outcome outcome = await sLink.SendMessageAsync(message);
+                        Assert.True(outcome.DescriptorCode == Accepted.Code, "message is not accepted.");
+                    }).ToArray());
+                }
+            });
+
+            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);
+                        if (Interlocked.Increment(ref count) > 10000)
+                        {
+                            break;
+                        }
+                    }
+                });
+            }).ToArray());
+
+            done = true;
+            sendTask.Wait();
+
+            connection.Close();
+        }
+
         [Fact]
         public void AmqpSequenceNumberWrapAroundTest()
         {