Exception from cancellation when a token is already cancelled.

Azure · Aug 20, 2021 · 09f1a60 · 09f1a60
1 parent c224ae1
commit 09f1a60
Show file tree

Hide file tree

Showing 8 changed files with 281 additions and 58 deletions.
diff --git a/Microsoft.Azure.Amqp/Amqp/AmqpObject.cs b/Microsoft.Azure.Amqp/Amqp/AmqpObject.cs
@@ -595,8 +595,6 @@ public void Signal(bool syncComplete, Exception exception)
 
             protected void Start()
             {
-                this.SetTimer();
-
                 bool shouldComplete = false;
                 Exception completeException = null;
 
@@ -618,6 +616,8 @@ protected void Start()
                     amqpObject.pendingClose = null;
                     this.Signal(true, completeException);
                 }
+
+                this.StartTracking();
             }
 
             protected abstract bool OnStart();

diff --git a/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs
@@ -874,7 +874,6 @@ public static Outcome End(IAsyncResult result)
 
             public void Start()
             {
-                this.SetTimer();
                 DeliveryState deliveryState;
                 if (txnId.Array != null)
                 {
@@ -894,6 +893,8 @@ public void Start()
                     // Delivery tag not found
                     link.pendingDispositions.CompleteWork(deliveryTag, true, AmqpConstants.RejectedNotFoundOutcome);
                 }
+
+                this.StartTracking();
             }
 
             public void Done(bool completedSynchronously, DeliveryState state)

diff --git a/Microsoft.Azure.Amqp/Amqp/RequestResponseAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/RequestResponseAmqpLink.cs
@@ -300,9 +300,9 @@ public static AmqpMessage End(IAsyncResult result)
 
             public void Start()
             {
-                this.SetTimer();
                 this.parent.sender.SendMessageNoWait(this.request, AmqpConstants.EmptyBinary, this.transactionId);
                 this.request = null;
+                this.StartTracking();
             }
 
             public void Done(bool completedSynchronously, AmqpMessage response)

diff --git a/Microsoft.Azure.Amqp/Amqp/SendingAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/SendingAmqpLink.cs
@@ -296,8 +296,8 @@ public static Outcome End(IAsyncResult result)
 
             public void Start()
             {
-                this.SetTimer();
                 this.link.SendMessageInternal(this.message, this.deliveryTag, this.txnId);
+                this.StartTracking();
             }
 
             public void Done(bool completedSynchronously, Outcome outcome)

diff --git a/Microsoft.Azure.Amqp/Amqp/TimeoutAsyncResult.cs b/Microsoft.Azure.Amqp/Amqp/TimeoutAsyncResult.cs
@@ -9,37 +9,36 @@ namespace Microsoft.Azure.Amqp
     abstract class TimeoutAsyncResult<T> : AsyncResult where T : class
     {
         readonly TimeSpan timeout;
-        readonly CancellationTokenRegistration cancellationTokenRegistration;
+        readonly CancellationToken cancellationToken;
+        CancellationTokenRegistration cancellationTokenRegistration;
         Timer timer;
-#if DEBUG
-        bool setTimerCalled;  // make sure derived class always call SetTimer
-#endif
 
         protected TimeoutAsyncResult(TimeSpan timeout, CancellationToken cancellationToken, AsyncCallback callback, object state)
             : base(callback, state)
         {
-            // The derived class must call SetTimer to start the timer.
-            // Timer is not started here because it could fire before the
-            // derived class ctor completes.
+            // The derived class must call StartTracking to start the timer and cancelation token registration.
+            // Timer is not started here because it could fire before the derived class ctor completes.
             this.timeout = timeout;
-            if (cancellationToken.CanBeCanceled)
-            {
-                this.cancellationTokenRegistration = cancellationToken.Register(o => ((TimeoutAsyncResult<T>)o).Cancel(), this);
-            }
+            this.cancellationToken = cancellationToken;
         }
 
         protected abstract T Target { get; }
 
         public abstract void Cancel();
 
-        protected void SetTimer()
+        protected void StartTracking()
         {
-#if DEBUG
-            this.setTimerCalled = true;
-#endif
-            if (this.timeout != TimeSpan.MaxValue)
+            if (!this.IsCompleted)
             {
-                this.timer = new Timer(s => OnTimerCallback(s), this, this.timeout, Timeout.InfiniteTimeSpan);
+                if (this.timeout != Timeout.InfiniteTimeSpan && this.timeout != TimeSpan.MaxValue)
+                {
+                    this.timer = new Timer(s => OnTimerCallback(s), this, this.timeout, Timeout.InfiniteTimeSpan);
+                }
+
+                if (this.cancellationToken.CanBeCanceled)
+                {
+                    this.cancellationTokenRegistration = this.cancellationToken.Register(o => ((TimeoutAsyncResult<T>)o).Cancel(), this);
+                }
             }
         }
 
@@ -78,9 +77,6 @@ static void OnTimerCallback(object state)
 
         bool CompleteInternal(bool syncComplete, Exception exception)
         {
-#if DEBUG
-            Fx.AssertAndThrow(exception != null || this.setTimerCalled, "Must call SetTimer.");
-#endif
             if (this.timer != null)
             {
                 this.timer.Dispose();

diff --git a/Microsoft.Azure.Amqp/Amqp/Transport/AmqpTransportInitiator.cs b/Microsoft.Azure.Amqp/Amqp/Transport/AmqpTransportInitiator.cs
@@ -265,12 +265,12 @@ public ConnectAsyncResult(AmqpTransportInitiator initiator, TimeSpan timeout, Ca
                 this.args = new TransportAsyncCallbackArgs();
                 this.args.CompletedCallback = onConnect;
                 this.args.UserToken = this;
-                this.SetTimer();
-
                 if (!initiator.ConnectAsync(timeout, this.args))
                 {
                     OnConnect(this.args);
                 }
+
+                this.StartTracking();
             }
 
             protected override string Target

diff --git a/Microsoft.Azure.Amqp/Amqp/Transport/AmqpTransportProvider.cs b/Microsoft.Azure.Amqp/Amqp/Transport/AmqpTransportProvider.cs
@@ -10,6 +10,12 @@ public AmqpTransportProvider()
             this.ProtocolId = ProtocolId.Amqp;
         }
 
+        internal AmqpTransportProvider(AmqpVersion version)
+            : this()
+        {
+            this.Versions.Add(version);
+        }
+
         protected override TransportBase OnCreateTransport(TransportBase innerTransport, bool isInitiator)
         {
             return innerTransport;