[#77] throw exception in send/receive if the object is closed.

Azure · Feb 12, 2018 · d6e42f8 · d6e42f8
1 parent c042f90
commit d6e42f8
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 2 deletions.
diff --git a/Microsoft.Azure.Amqp/Amqp/AmqpObject.cs b/Microsoft.Azure.Amqp/Amqp/AmqpObject.cs
@@ -482,6 +482,14 @@ protected void OnReceiveCloseCommand(string command, Error error)
             }
         }
 
+        protected void ThrowIfClosed()
+        {
+            if (this.closeCalled)
+            {
+                throw new AmqpException(AmqpErrorCode.IllegalState, $"'{this.name}' is closed");
+            }
+        }
+
         static void OnSafeCloseComplete(IAsyncResult result)
         {
             AmqpObject thisPtr = (AmqpObject)result.AsyncState;

diff --git a/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/ReceivingAmqpLink.cs
@@ -36,7 +36,7 @@ public ReceivingAmqpLink(AmqpLinkSettings settings)
         }
 
         public ReceivingAmqpLink(AmqpSession session, AmqpLinkSettings settings) :
-            base(session, settings)
+            base("receiver", session, settings)
         {
         }
 
@@ -157,6 +157,7 @@ public IAsyncResult BeginReceiveMessages(int messageCount, TimeSpan timeout, Asy
 
         IAsyncResult BeginReceiveMessages(int messageCount, TimeSpan batchWaitTimeout, TimeSpan timeout, AsyncCallback callback, object state)
         {
+            this.ThrowIfClosed();
             List<AmqpMessage> messages = new List<AmqpMessage>();
             lock (this.SyncRoot)
             {
@@ -249,6 +250,7 @@ public Task<Outcome> DisposeMessageAsync(ArraySegment<byte> deliveryTag, Outcome
 
         public IAsyncResult BeginDisposeMessage(ArraySegment<byte> deliveryTag, Outcome outcome, bool batchable, TimeSpan timeout, AsyncCallback callback, object state)
         {
+            this.ThrowIfClosed();
             return new DisposeAsyncResult(this, deliveryTag, outcome, batchable, timeout, callback, state);
         }
 
@@ -293,6 +295,7 @@ public void ModifyMessage(AmqpMessage message, bool deliveryFailed, bool deliver
 
         public void DisposeMessage(AmqpMessage message, DeliveryState state, bool settled, bool batchable)
         {
+            this.ThrowIfClosed();
             message.Batchable = batchable;
             this.DisposeDelivery(message, settled, state);
         }

diff --git a/Microsoft.Azure.Amqp/Amqp/SendingAmqpLink.cs b/Microsoft.Azure.Amqp/Amqp/SendingAmqpLink.cs
@@ -24,7 +24,7 @@ public SendingAmqpLink(AmqpLinkSettings settings)
         }
 
         public SendingAmqpLink(AmqpSession session, AmqpLinkSettings settings)
-            : base(session, settings)
+            : base("sender", session, settings)
         {
             // TODO: Add capability negotiation logic for BatchedMessageFormat to this.Settings
             this.pendingDeliveries = new SerializedWorker<AmqpMessage>(this);
@@ -76,6 +76,7 @@ public IAsyncResult BeginSendMessage(
             AsyncCallback callback, 
             object state)
         {
+            this.ThrowIfClosed();
             if (this.dispositionListener != null)
             {
                 throw new InvalidOperationException(CommonResources.DispositionListenerSetNotSupported);