public HttpRestClientConfiguration(HttpMessageHandler defaultHttpClientHandler = null,
            IEnumerable<DelegatingHandler> messageProcessingHandlers = null,
            IRetryPolicy retryPolicy = null, 
            TimeSpan? httpRequestTimeout = null)
        {
            this._httpRequestTimeout = httpRequestTimeout ?? TimeSpan.FromMinutes(1);
            this.HttpMessageHandler = defaultHttpClientHandler ?? new HttpClientHandler();
            if (messageProcessingHandlers == null)
            {
                this._additionalDelegatingHandlers = new ReadOnlyCollection<DelegatingHandler>(new List<DelegatingHandler>());
            }
            else
            {
                this._additionalDelegatingHandlers = new ReadOnlyCollection<DelegatingHandler>(messageProcessingHandlers.ToList());
            }

            this.RetryPolicy = retryPolicy ?? new NoRetryPolicy();

            //Here we chain the delegating handlers such that each delegating handler has the next one as its inner handler
            for (int i = this.DelegatingHandlers.Count - 1; i >= 0; i--)
            {
                if (i - 1 >= 0)
                {
                    this.DelegatingHandlers[i - 1].InnerHandler = this.DelegatingHandlers[i];
                }
            }

            //the first delegating handler has the default defaultHttpClientHandler as its inner handler
            if (this.DelegatingHandlers.Any())
            {
                this.DelegatingHandlers.Last().InnerHandler = this.HttpMessageHandler;
            }

        }

        public static void Execute(Action method, IRetryPolicy retryPolicy, IBackOffScheme backOffScheme, int numRetries = Constants.LoadBalancingHelperNumRetriesDefault)
        {
            int retryCount = 0;
            bool requestSuccess = false;
            while ((!requestSuccess) && (retryCount < numRetries))
            {
                try
                {
                    method();
                    requestSuccess = true;
                }
                catch (Exception ex)
                {
                    Trace.TraceError("\tAttempt {0} failed with exception {1} - {2}", retryCount, ex.GetType(), ex.Message);

                    retryCount++;
                    if ((retryCount < numRetries) && (retryPolicy.ShouldRetryAttempt(ex)))
                    {
                        var sleepInterval = backOffScheme.GetRetryInterval(retryCount);
                        if (sleepInterval != default (TimeSpan))
                        {
                            Trace.TraceInformation("\tWill retry after {0} milliseconds......", sleepInterval.TotalMilliseconds);
                            Thread.Sleep(sleepInterval);
                        }
                    }
                    else
                    {
                        throw;
                    }
                }
            }
        }

        public Task DownloadBlob(
            Uri uri,
            string localFile,
            FileEncryption fileEncryption,
            ulong initializationVector,
            CloudBlobClient client,
            CancellationToken cancellationToken,
            IRetryPolicy retryPolicy,
            Func<string> getSharedAccessSignature = null,
            long start = 0,
            long length = -1,
            int parallelTransferThreadCount = 10,
            int numberOfConcurrentTransfers = 2)
        {
            if (client != null && getSharedAccessSignature != null)
            {
                throw new InvalidOperationException("The arguments client and getSharedAccessSignature cannot both be non-null");
            }

            SetConnectionLimits(uri, Environment.ProcessorCount * numberOfConcurrentTransfers * parallelTransferThreadCount);

            Task task =
                Task.Factory.StartNew(
                    () =>
                        DownloadFileFromBlob(uri, localFile, fileEncryption, initializationVector, client,
                            cancellationToken, retryPolicy, getSharedAccessSignature, start: start, length: length,
                            parallelTransferThreadCount: parallelTransferThreadCount));
            return task;
        }

 public static Task<SqlDataReader> ExecuteReaderAsyncWithRetry(
                                                               SqlCommand command,
                                                               CommandBehavior behavior,
                                                               IRetryPolicy retryPolicy)
 {
     return retryPolicy.ExecuteAsyncWithRetry(() => command.ExecuteReaderAsync(behavior));
 }

 public static Task<int> ExecuteNonQueryAsyncWithRetry(
                                                       SqlCommand command,
                                                       CancellationToken cancellationToken,
                                                       IRetryPolicy retryPolicy)
 {
     return retryPolicy.ExecuteAsyncWithRetry(() => command.ExecuteNonQueryAsync(cancellationToken));
 }

 public Task UploadBlob(
     Uri url,
     string localFile,
     FileEncryption fileEncryption,
     CancellationToken cancellationToken,
     CloudBlobClient client,
     IRetryPolicy retryPolicy,
     string contentType = null,
     string subDirectory = "",
     Func<string> getSharedAccessSignature = null)
 {
     SetConnectionLimits(url);
     return Task.Factory.StartNew(
         () => UploadFileToBlob(
             cancellationToken,
             url,
             localFile,
             contentType,
             subDirectory,
             fileEncryption,
             client,
             retryPolicy,
             getSharedAccessSignature),
         cancellationToken);
 }

        /// <summary>
        /// Initialize a <see cref="HaConnection"/> with a list of <see cref="ManagedConnectionFactory"/>
        /// These connection factories are responsibile for creating <see cref="IConnection"/> to nodes in the clusters
        /// </summary>
        /// <param name="retryPolicy"></param>
        /// <param name="watcher"></param>
        /// <param name="connectionFactories"></param>
        public HaConnection(IRetryPolicy retryPolicy, IRabbitWatcher watcher, IList<ManagedConnectionFactory> connectionFactories) : base(retryPolicy, watcher)
        {        
            _connectionFactories = new RoundRobinList<ConnectionFactory>(connectionFactories);
            ConnectionEstablished handler = (endpoint, virtualHost) =>
            {
                if (_connectionFactories.All.Any(f => f.Endpoint + f.VirtualHost == endpoint + virtualHost))
                {
                    if (!IsConnected)
                    {
                        while (_connectionFactories.Current.Endpoint + _connectionFactories.Current.VirtualHost != endpoint + virtualHost)
                        {
                            //IF there are 2 different Tunnels using 2 HaConnection with 2 lists of cluster nodes in different orders:
                            //Example:

                            // ConnectionString1: host=q1;username=guest;password=guest|host=q2;username=guest;password=guest|host=q3;username=guest;password=guest
                            // ConnectionString2: host=q2;username=guest;password=guest|host=q3;username=guest;password=guest|host=q1;username=guest;password=guest

                            // When the first tunnel established the connection successfully to q1, it fires event and these lines of code is triggered.
                            // The 2nd HaConnection needs to set it's _connectionFactories.Current to q1 established by other tunnel. Before changing, _connectionFactories.Current is q3 by the order in the ConnectionString2 
                            _connectionFactories.GetNext();
                        }
                    }

                    FireConnectedEvent();
                }
            };
            ManagedConnectionFactory.ConnectionEstablished += handler;
            _unsubscribeEvents = () => { ManagedConnectionFactory.ConnectionEstablished -= handler; };
        }

 public static Task OpenAsyncWithRetry(
                                       this SqlConnection connection,
                                       CancellationToken cancellationToken,
                                       IRetryPolicy retryPolicy)
 {
     return retryPolicy.ExecuteAsyncWithRetry(() => connection.OpenAsync(cancellationToken));
 }

 public Task UploadBlob(
     Uri url,
     string localFile,
     FileEncryption fileEncryption,
     CancellationToken cancellationToken,
     CloudBlobClient client,
     IRetryPolicy retryPolicy,
     string contentType = null,
     string subDirectory = "",
     Func<string> getSharedAccessSignature = null,
     int parallelTransferThreadCount = 10,
     int numberOfConcurrentTransfers = default(int))
 {
     SetConnectionLimits(url, numberOfConcurrentTransfers);
     return Task.Factory.StartNew(
         () => UploadFileToBlob(
             cancellationToken,
             url,
             localFile,
             contentType,
             subDirectory,
             fileEncryption,
             client,
             retryPolicy,
             getSharedAccessSignature,
             parallelTransferThreadCount),
             cancellationToken);
 }

        public EventStreamProducer(IEventStreamWriter streamWriter, IRetryPolicy retryPolicy)
        {
            Require.NotNull(streamWriter, "streamWriter");
            Require.NotNull(retryPolicy, "retryPolicy");

            m_streamWriter = streamWriter;
            m_retryPolicy = retryPolicy;
        }

 /// <summary>
 ///  Creates a new <code>Policies</code> object using the provided policies.
 /// </summary>
 /// <param name="loadBalancingPolicy"> the load balancing policy to use. </param>
 /// <param name="reconnectionPolicy"> the reconnection policy to use. </param>
 /// <param name="retryPolicy"> the retry policy to use.</param>
 public Policies(ILoadBalancingPolicy loadBalancingPolicy,
                 IReconnectionPolicy reconnectionPolicy,
                 IRetryPolicy retryPolicy)
 {
     this._loadBalancingPolicy = loadBalancingPolicy;
     this._reconnectionPolicy = reconnectionPolicy;
     this._retryPolicy = retryPolicy;
 }

 /// <summary>
 /// Initializes a new instance of the AbstractionContext class.
 /// </summary>
 /// <param name="tokenSource">A Cancellation token source.</param>
 /// <param name="logger">A logger instance.</param>
 /// <param name="httpOperationTimeout">The HTTP operation timeout.</param>
 /// <param name="retryPolicy">The retry policy.</param>
 public AbstractionContext(CancellationTokenSource tokenSource, ILogger logger, TimeSpan httpOperationTimeout, IRetryPolicy retryPolicy)
 {
     this.RetryPolicy = retryPolicy;
     tokenSource.ArgumentNotNull("tokenSource");
     logger.ArgumentNotNull("logger");
     this.CancellationTokenSource = tokenSource;
     this.Logger = logger;
     this.HttpOperationTimeout = httpOperationTimeout;
 }

        public HttpRestClientRetryPolicy(IRetryPolicy retryPolicy)
        {
            if (retryPolicy == null)
            {
                throw new ArgumentNullException("retryPolicy");
            }

            this.retryPolicy = retryPolicy;
        }

        public ElasticRequestProcessor(IElasticConnection connection, IElasticMapping mapping, ILog log, IRetryPolicy retryPolicy)
        {
            Argument.EnsureNotNull("connection", connection);
            Argument.EnsureNotNull("mapping", mapping);
            Argument.EnsureNotNull("log", log);
            Argument.EnsureNotNull("retryPolicy", retryPolicy);

            this.connection = connection;
            this.mapping = mapping;
            this.log = log;
            this.retryPolicy = retryPolicy;
        }

 internal DurableConnection(IRetryPolicy retryPolicy, IRabbitWatcher watcher)
 {
     if (retryPolicy == null)
     {
         throw new ArgumentNullException(nameof(retryPolicy));
     }
     if (watcher == null)
     {
         throw new ArgumentNullException(nameof(watcher));
     }
     _retryPolicy = retryPolicy;
     _watcher = watcher;
 }

 /**
  *
  * @param connectString list of servers to connect to
  * @param sessionTimeoutMs session timeout
  * @param connectionTimeoutMs connection timeout
  * @param watcher default watcher or null
  * @param retryPolicy the retry policy to use
  */
 public CuratorZookeeperClient(String connectString, 
                                 int sessionTimeoutMs, 
                                 int connectionTimeoutMs, 
                                 Watcher watcher, 
                                 IRetryPolicy retryPolicy)
     : this(new DefaultZookeeperFactory(),
             new FixedEnsembleProvider(connectString),
             sessionTimeoutMs,
             connectionTimeoutMs,
             watcher,
             retryPolicy,
             false)
 {
 }

		/**
     * @param zookeeperFactory factory for creating {@link ZooKeeper} instances
     * @param ensembleProvider the ensemble provider
     * @param sessionTimeoutMs session timeout
     * @param connectionTimeoutMs connection timeout
     * @param watcher default watcher or null
     * @param retryPolicy the retry policy to use
     * @param canBeReadOnly if true, allow ZooKeeper client to enter
     *                      read only mode in case of a network partition. See
     *                      {@link ZooKeeper#ZooKeeper(String, int, Watcher, long, byte[], bool)}
     *                      for details
     */
		public CuratorZookeeperClient(IZookeeperFactory zookeeperFactory, IEnsembleProvider ensembleProvider, int sessionTimeoutMs, int connectionTimeoutMs, IWatcher watcher, IRetryPolicy retryPolicy, bool canBeReadOnly)
		{
			if ( sessionTimeoutMs < connectionTimeoutMs )
			{
				log.Warn(String.Format("session timeout [{0}] is less than connection timeout [{1}]", sessionTimeoutMs, connectionTimeoutMs));
			}


//			retryPolicy = Preconditions.checkNotNull(retryPolicy, "retryPolicy cannot be null");
//			ensembleProvider = Preconditions.checkNotNull(ensembleProvider, "ensembleProvider cannot be null");

			this.connectionTimeoutMs = connectionTimeoutMs;
			state = new ConnectionState(zookeeperFactory, ensembleProvider, TimeSpan.FromMilliseconds(sessionTimeoutMs), TimeSpan.FromMilliseconds(connectionTimeoutMs), watcher, tracer, canBeReadOnly);
			SetRetryPolicy(retryPolicy);
		}

        private static async Task TestSqlRetryPolicy(IRetryPolicy retryPolicy, int interval)
        {
            var retryCount = 0;
            retryPolicy.Retry += (sender, args) =>
            {
                retryCount++;

                // Assert
                Assert.Equal(typeof(TimeoutException), args.Exception.GetType());
                Assert.Equal(retryCount, args.RetryCount);
                Assert.Equal(retryCount * interval, args.Delay.TotalMilliseconds);
            };

            var taskFunction = TaskFunctionTestFactory.GetTaskFunctionTResultWithRetry();

            // Act
            await retryPolicy.ExecuteAsyncWithRetry(taskFunction);
        }

        public DurableConnection(IRetryPolicy retryPolicy, IRabbitWatcher watcher, ConnectionFactory connectionFactory)
        {
            if (retryPolicy == null)
            {
                throw new ArgumentNullException("retryPolicy");
            }
            if (watcher == null)
            {
                throw new ArgumentNullException("watcher");
            }
            if (connectionFactory == null)
            {
                throw new ArgumentNullException("connectionFactory");
            }

            _retryPolicy = retryPolicy;
            _watcher = watcher;
            ConnectionFactory = connectionFactory;
        }

        public Task DownloadBlob(
            Uri uri, 
            string localFile, 
            FileEncryption fileEncryption, 
            ulong initializationVector, 
            CloudBlobClient client, 
            CancellationToken cancellationToken, 
            IRetryPolicy retryPolicy, 
            Func<string> getSharedAccessSignature = null,
            long start = 0,
            long length = -1)
        {
            if (client != null && getSharedAccessSignature != null)
            {
                throw new InvalidOperationException("The arguments client and getSharedAccessSignature cannot both be non-null");
            }

            SetConnectionLimits(uri);

            Task task = Task.Factory.StartNew(() => DownloadFileFromBlob(uri, localFile, fileEncryption, initializationVector, client, cancellationToken, retryPolicy, getSharedAccessSignature, start:start, length:length));
            return task;
        }