Retrypolicy (Azure#110)

Implemented an abstract RetryPolicy class with two implementations: NoRetry and RetryExponential
(Issue Azure#7)

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

@@ -13,14 +13,22 @@ class AmqpClient : IInnerSenderReceiver
         MessageSender innerSender;
         MessageReceiver innerReceiver;
 
-        internal AmqpClient(ServiceBusConnection servicebusConnection, string entityPath, MessagingEntityType entityType, ReceiveMode mode = ReceiveMode.ReceiveAndDelete)
+        internal AmqpClient(
+            ServiceBusConnection servicebusConnection,
+            string entityPath,
+            MessagingEntityType entityType,
+            RetryPolicy retryPolicy,
+            ReceiveMode mode = ReceiveMode.ReceiveAndDelete)
         {
             this.ServiceBusConnection = servicebusConnection;
             this.EntityPath = entityPath;
             this.MessagingEntityType = entityType;
             this.ReceiveMode = mode;
-            this.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(servicebusConnection.SasKeyName, servicebusConnection.SasKey);
+            this.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(
+                servicebusConnection.SasKeyName,
+                servicebusConnection.SasKey);
             this.CbsTokenProvider = new TokenProviderAdapter(this.TokenProvider, servicebusConnection.OperationTimeout);
+            this.RetryPolicy = retryPolicy;
         }
 
         public MessageSender InnerSender
@@ -71,6 +79,8 @@ public MessageReceiver InnerReceiver
 
         internal ICbsTokenProvider CbsTokenProvider { get; }
 
+        internal RetryPolicy RetryPolicy { get; }
+
         protected object ThisLock { get; } = new object();
 
         TokenProvider TokenProvider { get; }
@@ -85,12 +95,24 @@ public async Task CloseAsync()
 
         MessageSender CreateMessageSender()
         {
-            return new AmqpMessageSender(this.EntityPath, this.MessagingEntityType, this.ServiceBusConnection, this.CbsTokenProvider);
+            return new AmqpMessageSender(
+                this.EntityPath,
+                this.MessagingEntityType,
+                this.ServiceBusConnection,
+                this.CbsTokenProvider,
+                this.RetryPolicy);
         }
 
         MessageReceiver CreateMessageReceiver()
         {
-            return new AmqpMessageReceiver(this.EntityPath, this.MessagingEntityType, this.ReceiveMode, this.ServiceBusConnection.PrefetchCount, this.ServiceBusConnection, this.CbsTokenProvider);
+            return new AmqpMessageReceiver(
+                this.EntityPath,
+                this.MessagingEntityType,
+                this.ReceiveMode,
+                this.ServiceBusConnection.PrefetchCount,
+                this.ServiceBusConnection,
+                this.CbsTokenProvider,
+                this.RetryPolicy);
         }
     }
 }

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

@@ -25,13 +25,29 @@ sealed class AmqpMessageReceiver : MessageReceiver
         string sessionId;
         DateTime lockedUntilUtc;
 
-        internal AmqpMessageReceiver(string entityName, MessagingEntityType? entityType, ReceiveMode mode, int prefetchCount, ServiceBusConnection serviceBusConnection, ICbsTokenProvider cbsTokenProvider)
-            : this(entityName, entityType, mode, prefetchCount, serviceBusConnection, cbsTokenProvider, null)
+        internal AmqpMessageReceiver(
+            string entityName,
+            MessagingEntityType? entityType,
+            ReceiveMode mode,
+            int prefetchCount,
+            ServiceBusConnection serviceBusConnection,
+            ICbsTokenProvider cbsTokenProvider,
+            RetryPolicy retryPolicy)
+            : this(entityName, entityType, mode, prefetchCount, serviceBusConnection, cbsTokenProvider, null, retryPolicy)
         {
         }
 
-        internal AmqpMessageReceiver(string entityName, MessagingEntityType? entityType, ReceiveMode mode, int prefetchCount, ServiceBusConnection serviceBusConnection, ICbsTokenProvider cbsTokenProvider, string sessionId, bool isSessionReceiver = false)
-            : base(mode, serviceBusConnection.OperationTimeout)
+        internal AmqpMessageReceiver(
+            string entityName,
+            MessagingEntityType? entityType,
+            ReceiveMode mode,
+            int prefetchCount,
+            ServiceBusConnection serviceBusConnection,
+            ICbsTokenProvider cbsTokenProvider,
+            string sessionId,
+            RetryPolicy retryPolicy,
+            bool isSessionReceiver = false)
+            : base(mode, serviceBusConnection.OperationTimeout, retryPolicy)
         {
             this.entityName = entityName;
             this.EntityType = entityType;

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

@@ -18,8 +18,8 @@ sealed class AmqpMessageSender : MessageSender
     {
         int deliveryCount;
 
-        internal AmqpMessageSender(string entityName, MessagingEntityType? entityType, ServiceBusConnection serviceBusConnection, ICbsTokenProvider cbsTokenProvider)
-            : base(serviceBusConnection.OperationTimeout)
+        internal AmqpMessageSender(string entityName, MessagingEntityType? entityType, ServiceBusConnection serviceBusConnection, ICbsTokenProvider cbsTokenProvider, RetryPolicy retryPolicy)
+            : base(serviceBusConnection.OperationTimeout, retryPolicy)
         {
             this.Path = entityName;
             this.EntityType = entityType;

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

@@ -12,8 +12,8 @@ namespace Microsoft.Azure.ServiceBus.Amqp
 
     class AmqpMessageSession : MessageSession
     {
-        public AmqpMessageSession(string sessionId, DateTime lockedUntilUtc, MessageReceiver innerMessageReceiver)
-            : base(innerMessageReceiver.ReceiveMode, sessionId, lockedUntilUtc, innerMessageReceiver)
+        public AmqpMessageSession(string sessionId, DateTime lockedUntilUtc, MessageReceiver innerMessageReceiver, RetryPolicy retryPolicy)
+            : base(innerMessageReceiver.ReceiveMode, sessionId, lockedUntilUtc, innerMessageReceiver, retryPolicy)
         {
         }
 

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

@@ -16,8 +16,9 @@ public AmqpSubscriptionClient(
             ServiceBusConnection servicebusConnection,
             string entityPath,
             MessagingEntityType entityType,
+            RetryPolicy retryPolicy,
             ReceiveMode mode = ReceiveMode.ReceiveAndDelete)
-            : base(servicebusConnection, entityPath, entityType, mode)
+            : base(servicebusConnection, entityPath, entityType, retryPolicy, mode)
         {
         }
 

src/Microsoft.Azure.ServiceBus/Constants.cs

@@ -34,5 +34,7 @@ static class Constants
         public static readonly TimeSpan MinimumLockDuration = TimeSpan.FromSeconds(5);
 
         public static readonly TimeSpan MaximumRenewBufferDuration = TimeSpan.FromSeconds(10);
+
+        public static readonly TimeSpan DefaultRetryDeltaBackoff = TimeSpan.FromSeconds(3);
     }
 }

src/Microsoft.Azure.ServiceBus/Core/MessageReceiver.cs

@@ -18,8 +18,8 @@ internal abstract class MessageReceiver : ClientEntity, IMessageReceiver
         MessageReceivePump receivePump;
         CancellationTokenSource receivePumpCancellationTokenSource;
 
-        protected MessageReceiver(ReceiveMode receiveMode, TimeSpan operationTimeout)
-            : base(nameof(MessageReceiver) + StringUtility.GetRandomString())
+        protected MessageReceiver(ReceiveMode receiveMode, TimeSpan operationTimeout, RetryPolicy retryPolicy)
+            : base(nameof(MessageReceiver) + StringUtility.GetRandomString(), retryPolicy ?? RetryPolicy.Default)
         {
             this.ReceiveMode = receiveMode;
             this.operationTimeout = operationTimeout;
@@ -133,10 +133,15 @@ public async Task<IList<Message>> ReceiveAsync(int maxMessageCount, TimeSpan ser
         {
             MessagingEventSource.Log.MessageReceiveStart(this.ClientId, maxMessageCount);
 
-            IList<Message> messages;
+            IList<Message> messages = null;
             try
             {
-                messages = await this.OnReceiveAsync(maxMessageCount, serverWaitTime).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        messages = await this.OnReceiveAsync(maxMessageCount, serverWaitTime).ConfigureAwait(false);
+                    }, serverWaitTime)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -166,10 +171,15 @@ public async Task<IList<Message>> ReceiveBySequenceNumberAsync(IEnumerable<long>
 
             MessagingEventSource.Log.MessageReceiveBySequenceNumberStart(this.ClientId, count, sequenceNumbers);
 
-            IList<Message> messages;
+            IList<Message> messages = null;
             try
             {
-                messages = await this.OnReceiveBySequenceNumberAsync(sequenceNumbers).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        messages = await this.OnReceiveBySequenceNumberAsync(sequenceNumbers).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -196,7 +206,12 @@ public async Task CompleteAsync(IEnumerable<string> lockTokens)
 
             try
             {
-                await this.OnCompleteAsync(lockTokens).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnCompleteAsync(lockTokens).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -214,7 +229,12 @@ public async Task AbandonAsync(string lockToken)
             MessagingEventSource.Log.MessageAbandonStart(this.ClientId, 1, lockToken);
             try
             {
-                await this.OnAbandonAsync(lockToken).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnAbandonAsync(lockToken).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -233,7 +253,12 @@ public async Task DeferAsync(string lockToken)
 
             try
             {
-                await this.OnDeferAsync(lockToken).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnDeferAsync(lockToken).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -252,7 +277,12 @@ public async Task DeadLetterAsync(string lockToken)
 
             try
             {
-                await this.OnDeadLetterAsync(lockToken).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnDeadLetterAsync(lockToken).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -269,10 +299,15 @@ public async Task<DateTime> RenewLockAsync(string lockToken)
 
             MessagingEventSource.Log.MessageRenewLockStart(this.ClientId, 1, lockToken);
 
-            DateTime lockedUntilUtc;
+            DateTime lockedUntilUtc = DateTime.Now;
             try
             {
-                lockedUntilUtc = await this.OnRenewLockAsync(lockToken).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        lockedUntilUtc = await this.OnRenewLockAsync(lockToken).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -320,12 +355,17 @@ public async Task<Message> PeekBySequenceNumberAsync(long fromSequenceNumber)
         /// <returns>A batch of messages peeked.</returns>
         public async Task<IList<Message>> PeekBySequenceNumberAsync(long fromSequenceNumber, int messageCount)
         {
-            IList<Message> messages;
+            IList<Message> messages = null;
 
             MessagingEventSource.Log.MessagePeekStart(this.ClientId, fromSequenceNumber, messageCount);
             try
             {
-                messages = await this.OnPeekAsync(fromSequenceNumber, messageCount).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        messages = await this.OnPeekAsync(fromSequenceNumber, messageCount).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {

src/Microsoft.Azure.ServiceBus/Core/MessageSender.cs

@@ -13,8 +13,8 @@ internal abstract class MessageSender : ClientEntity, IMessageSender
              "StyleCop.CSharp.ReadabilityRules",
              "SA1126:PrefixCallsCorrectly",
              Justification = "This is not a method call, but a type.")]
-        protected MessageSender(TimeSpan operationTimeout)
-            : base(nameof(MessageSender) + StringUtility.GetRandomString())
+        protected MessageSender(TimeSpan operationTimeout, RetryPolicy retryPolicy)
+            : base(nameof(MessageSender) + StringUtility.GetRandomString(), retryPolicy ?? RetryPolicy.Default)
         {
             this.OperationTimeout = operationTimeout;
         }
@@ -35,7 +35,12 @@ public async Task SendAsync(IList<Message> messageList)
 
             try
             {
-                await this.OnSendAsync(messageList).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnSendAsync(messageList).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -64,11 +69,16 @@ public async Task<long> ScheduleMessageAsync(Message message, DateTimeOffset sch
             message.ScheduledEnqueueTimeUtc = scheduleEnqueueTimeUtc.UtcDateTime;
             MessageSender.ValidateMessage(message);
             MessagingEventSource.Log.ScheduleMessageStart(this.ClientId, scheduleEnqueueTimeUtc);
-            long result;
+            long result = 0;
 
             try
             {
-                result = await this.OnScheduleMessageAsync(message).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        result = await this.OnScheduleMessageAsync(message).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -86,7 +96,12 @@ public async Task CancelScheduledMessageAsync(long sequenceNumber)
 
             try
             {
-                await this.OnCancelScheduledMessageAsync(sequenceNumber).ConfigureAwait(false);
+                await this.RetryPolicy.RunOperation(
+                    async () =>
+                    {
+                        await this.OnCancelScheduledMessageAsync(sequenceNumber).ConfigureAwait(false);
+                    }, this.OperationTimeout)
+                    .ConfigureAwait(false);
             }
             catch (Exception exception)
             {

src/Microsoft.Azure.ServiceBus/MessageSession.cs

@@ -12,8 +12,8 @@ namespace Microsoft.Azure.ServiceBus
     internal abstract class MessageSession : MessageReceiver, IMessageSession
     {
         /// <summary>Represents a message session that allows grouping of related messages for processing in a single transaction.</summary>
-        protected MessageSession(ReceiveMode receiveMode, string sessionId, DateTime lockedUntilUtc, MessageReceiver innerReceiver)
-            : base(receiveMode, innerReceiver.OperationTimeout)
+        protected MessageSession(ReceiveMode receiveMode, string sessionId, DateTime lockedUntilUtc, MessageReceiver innerReceiver, RetryPolicy retryPolicy)
+            : base(receiveMode, innerReceiver.OperationTimeout, retryPolicy)
         {
             if (innerReceiver == null)
             {

src/Microsoft.Azure.ServiceBus/MessagingEventSource.cs

@@ -915,5 +915,20 @@ void MessageReceiverPumpRenewMessageException(string clientId, long sequenceNumb
         {
             this.WriteEvent(76, clientId, sequenceNumber, exception);
         }
+
+        [NonEvent]
+        public void RunOperationExceptionEncountered(Exception exception)
+        {
+            if (this.IsEnabled())
+            {
+                this.RunOperationExceptionEncountered(exception.ToString());
+            }
+        }
+
+        [Event(77, Level = EventLevel.Warning, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]
+        void RunOperationExceptionEncountered(string exception)
+        {
+            this.WriteEvent(77, exception);
+        }
     }
 }

src/Microsoft.Azure.ServiceBus/NoRetry.cs

@@ -0,0 +1,19 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Microsoft.Azure.ServiceBus
+{
+    using System;
+
+    public sealed class NoRetry : RetryPolicy
+    {
+        protected override bool OnShouldRetry(
+            TimeSpan remainingTime,
+            int currentRetryCount,
+            out TimeSpan retryInterval)
+        {
+            retryInterval = TimeSpan.Zero;
+            return false;
+        }
+    }
+}