Synnefo » Pithos MS Client

#region

/* -----------------------------------------------------------------------

 * <copyright file="CloudFilesClient.cs" company="GRNet">

 * 

 * Copyright 2011-2012 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 * </copyright>

 * -----------------------------------------------------------------------

 */

#endregion

// **CloudFilesClient** provides a simple client interface to CloudFiles and Pithos

//

// The class provides methods to upload/download files, delete files, manage containers

using System;

using System.Collections.Generic;

using System.Collections.Specialized;

using System.ComponentModel.Composition;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Net;

using System.Net.Http;

using System.Net.Http.Headers;

using System.Reflection;

using System.ServiceModel.Channels;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using Newtonsoft.Json;

using Newtonsoft.Json.Linq;

using Pithos.Interfaces;

using Pithos.Network;

using log4net;

namespace Pithos.Network

{

    [Export(typeof(ICloudClient))]

    public class CloudFilesClient:ICloudClient,IDisposable

    {

        private const string TOKEN_HEADER = "X-Auth-Token";

        private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);

        //CloudFilesClient uses *_baseClient* internally to communicate with the server

        //RestClient provides a REST-friendly interface over the standard WebClient.

        private RestClient _baseClient;

        private HttpClient _baseHttpClient;

        private HttpClient _baseHttpClientNoTimeout;

        //During authentication the client provides a UserName 

        public string UserName { get; set; }

        //and and ApiKey to the server

        public string ApiKey { get; set; }

        //And receives an authentication Token. This token must be provided in ALL other operations,

        //in the X-Auth-Token header

        private string _token;

        private readonly string _emptyGuid = Guid.Empty.ToString();

        private readonly Uri _emptyUri = new Uri("",UriKind.Relative);

        private HttpClientHandler _httpClientHandler = new HttpClientHandler

                                                           {

                                                               AllowAutoRedirect = true,

                                                               AutomaticDecompression = DecompressionMethods.GZip | DecompressionMethods.Deflate,

                                                               UseCookies = true,   

                                                           };

        public string Token

        {

            get { return _token; }

            set

            {

                _token = value;

                _baseClient.Headers[TOKEN_HEADER] = value;

            }

        }

        //The client also receives a StorageUrl after authentication. All subsequent operations must

        //use this url

        public Uri StorageUrl { get; set; }

        public Uri RootAddressUri

        {

            get { return _rootAddressUri; }

            set { _rootAddressUri = value; }

        }

        public double DownloadPercentLimit { get; set; }

        public double UploadPercentLimit { get; set; }

        public string AuthenticationUrl { get; set; }

        public string VersionPath

        {

            get { return UsePithos ? "v1" : "v1.0"; }

        }

        public bool UsePithos { get; set; }

        BufferManager _bufferManager=BufferManager.CreateBufferManager(TreeHash.DEFAULT_BLOCK_SIZE*4,(int)TreeHash.DEFAULT_BLOCK_SIZE);

        private string _userCatalogUrl;

        private Uri _rootAddressUri;

        public CloudFilesClient(string userName, string apiKey)

        {

            UserName = userName;

            ApiKey = apiKey;

            _userCatalogUrl = "https://pithos.okeanos.io/user_catalogs";

        }

        public CloudFilesClient(AccountInfo accountInfo)

        {

            Contract.Requires<ArgumentNullException>(accountInfo!=null,"accountInfo is null");

            Contract.Ensures(!String.IsNullOrWhiteSpace(Token));

            Contract.Ensures(StorageUrl != null);

            Contract.Ensures(_baseClient != null);

            Contract.Ensures(RootAddressUri != null);

            Contract.EndContractBlock();          

            _baseClient = new RestClient

            {

                BaseAddress = accountInfo.StorageUri.ToString(),

                Timeout = 30000,

                Retries = 3,

            };

            StorageUrl = accountInfo.StorageUri;

            Token = accountInfo.Token;

            UserName = accountInfo.UserName;

            _userCatalogUrl = "https://pithos.okeanos.io/user_catalogs";

            //Get the root address (StorageUrl without the account)

            var storageUrl = StorageUrl.AbsoluteUri;

            var usernameIndex = storageUrl.LastIndexOf(UserName);

            var rootUrl = storageUrl.Substring(0, usernameIndex);

            RootAddressUri = new Uri(rootUrl);

            //RootAddressUri = new Uri(storageUrl);

            var httpClientHandler = new HttpClientHandler

            {

                AllowAutoRedirect = true,

                AutomaticDecompression = DecompressionMethods.GZip | DecompressionMethods.Deflate,

                UseCookies = true,

            };

            _baseHttpClient = new HttpClient(httpClientHandler)

            {

                BaseAddress = StorageUrl,

                Timeout = TimeSpan.FromSeconds(30)

            };

            _baseHttpClient.DefaultRequestHeaders.Add(TOKEN_HEADER, Token);

            _baseHttpClientNoTimeout = new HttpClient(httpClientHandler)

            {

                BaseAddress = StorageUrl,

                Timeout = TimeSpan.FromMilliseconds(-1)

            };

            _baseHttpClientNoTimeout.DefaultRequestHeaders.Add(TOKEN_HEADER, Token);

        }

        private static void AssertStatusOK(HttpResponseMessage response, string message)

        {

            var statusCode = response.StatusCode;

            if (statusCode >= HttpStatusCode.BadRequest)

                throw new WebException(String.Format("{0} with code {1} - {2}", message, statusCode, response.ReasonPhrase));

        }

        public async Task<AccountInfo> Authenticate()

        {

            if (String.IsNullOrWhiteSpace(UserName))

                throw new InvalidOperationException("UserName is empty");

            if (String.IsNullOrWhiteSpace(ApiKey))

                throw new InvalidOperationException("ApiKey is empty");

            if (String.IsNullOrWhiteSpace(AuthenticationUrl))

                throw new InvalidOperationException("AuthenticationUrl is empty");

            //if (String.IsNullOrWhiteSpace(Token))

            //    throw new InvalidOperationException("Token is Empty");

            //Contract.Ensures(!String.IsNullOrWhiteSpace(Token));

            //Contract.Ensures(StorageUrl != null);

            //Contract.Ensures(_baseClient != null);

            //Contract.Ensures(RootAddressUri != null);

            //Contract.EndContractBlock();

            Log.InfoFormat("[AUTHENTICATE] Start for {0}", UserName);

            var groups = new List<Group>();

//            using (var authClient = new HttpClient(_httpClientHandler,false){ BaseAddress = new Uri(AuthenticationUrl),Timeout=TimeSpan.FromSeconds(30) })

            using (var authClient = new HttpClient(_httpClientHandler, false) { BaseAddress = new Uri(AuthenticationUrl), Timeout = TimeSpan.FromSeconds(30) })

            {                

                authClient.DefaultRequestHeaders.Add("X-Auth-User", UserName);

                authClient.DefaultRequestHeaders.Add("X-Auth-Key", ApiKey);

                string token;

                string objectStorageService;

                //using (var response = await authClient.GetAsyncWithRetries(new Uri(VersionPath, UriKind.Relative),3).ConfigureAwait(false)) // .DownloadStringWithRetryRelative(new Uri(VersionPath, UriKind.Relative), 3);                    

                using (var response = await authClient.SendAsyncWithRetries(new HttpRequestMessage(HttpMethod.Post,new Uri("", UriKind.Relative)), 3).ConfigureAwait(false))

                {

                    AssertStatusOK(response,"Authentication failed");

                    var body=await response.Content.ReadAsStringAsync();

                    JObject json = (JObject) JsonConvert.DeserializeObject(body);

                    dynamic jsonD = json;

                    objectStorageService = json["access"]["serviceCatalog"][6]["endpoints"][0]["publicURL"].ToString();

                    if (String.IsNullOrWhiteSpace(objectStorageService))

                        throw new InvalidOperationException("Failed to obtain storage url");

                    token = this.ApiKey;// response.Headers.GetFirstValue(TOKEN_HEADER);

                    if (String.IsNullOrWhiteSpace(token))

                        throw new InvalidOperationException("Failed to obtain token url");

                }

                string storageUrl = new Uri(objectStorageService).Combine(UserName).ToString();

                _baseClient = new RestClient

                {

                    BaseAddress = storageUrl,

                    Timeout = 30000,

                    Retries = 3,                    

                };

                StorageUrl = new Uri(storageUrl);

                Token = token;

                //Get the root address (StorageUrl without the account)

                var rootUrl = objectStorageService;//.Substring(0, usernameIndex);

                RootAddressUri = new Uri(rootUrl);

                _baseHttpClient = new HttpClient(_httpClientHandler,false)

                {

                    BaseAddress = StorageUrl,

                    Timeout = TimeSpan.FromSeconds(30)

                };

                _baseHttpClient.DefaultRequestHeaders.Add(TOKEN_HEADER, token);

                _baseHttpClientNoTimeout = new HttpClient(_httpClientHandler,false)

                {

                    BaseAddress = StorageUrl,

                    Timeout = TimeSpan.FromMilliseconds(-1)

                };

                _baseHttpClientNoTimeout.DefaultRequestHeaders.Add(TOKEN_HEADER, token);

                /* var keys = authClient.ResponseHeaders.AllKeys.AsQueryable();

                groups = (from key in keys

                            where key.StartsWith("X-Account-Group-")

                            let name = key.Substring(16)

                            select new Group(name, authClient.ResponseHeaders[key]))

                        .ToList();

*/

            }

            Log.InfoFormat("[AUTHENTICATE] End for {0}", UserName);

            Debug.Assert(_baseClient!=null);

            var displayName = UserName;

            Guid uuid;

            if (Guid.TryParse(UserName, out uuid))

            {

                displayName = await ResolveName(uuid);

            }

            return new AccountInfo {StorageUri = StorageUrl, Token = Token, UserName = UserName, DisplayName=displayName, Groups=groups};            

        }

        private static void TraceStart(string method, Uri actualAddress)

        {

            Log.InfoFormat("[{0}] {1} {2}", method, DateTime.Now, actualAddress);

        }

        private async Task<string> GetStringAsync(Uri targetUri, string errorMessage,DateTimeOffset? since=null)

        {

            TraceStart("GET",targetUri);

            var request = new HttpRequestMessage(HttpMethod.Get, targetUri);

            //request.Headers.Add("User-Agent", "Pithos+ Custom Header");

            if (since.HasValue)

            {

                request.Headers.IfModifiedSince = since.Value;

            }

            using (var response = await _baseHttpClient.SendAsyncWithRetries(request,3).ConfigureAwait(false))

            {

                AssertStatusOK(response, errorMessage);

                if (response.StatusCode == HttpStatusCode.NoContent)

                    return String.Empty;

                var content = await response.Content.ReadAsStringAsync().ConfigureAwait(false);

                return content;

            }

        }

        public async Task<IList<ContainerInfo>> ListContainers(string account)

        {

            var targetUrl = GetTargetUrl(account);

            var targetUri = new Uri(String.Format("{0}?format=json", targetUrl));

            var result = await GetStringAsync(targetUri, "List Containers failed").ConfigureAwait(false);

            if (String.IsNullOrWhiteSpace(result))

                return new List<ContainerInfo>();

            var infos = JsonConvert.DeserializeObject<IList<ContainerInfo>>(result);

            foreach (var info in infos)

            {

                info.Account = account;

            }

            return infos;

        }

        private string GetAccountUrl(string account)

        {

            return RootAddressUri.Combine(account).AbsoluteUri;

        }

        public IList<ShareAccountInfo> ListSharingAccounts(DateTime? since=null)

        {

            using (ThreadContext.Stacks["Share"].Push("List Accounts"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                var targetUri = new Uri(String.Format("{0}?format=json", RootAddressUri), UriKind.Absolute);

                var content=TaskEx.Run(async ()=>await GetStringAsync(targetUri, "ListSharingAccounts failed", since).ConfigureAwait(false)).Result;

                //If the result is empty, return an empty list,

                var infos = String.IsNullOrWhiteSpace(content)

                                ? new List<ShareAccountInfo>()

                            //Otherwise deserialize the account list into a list of ShareAccountInfos

                                : JsonConvert.DeserializeObject<IList<ShareAccountInfo>>(content);

                Log.DebugFormat("END");

                return infos;

            }

        }

        /// <summary>

        /// Request listing of all objects in a container modified since a specific time.

        /// If the *since* value is missing, return all objects

        /// </summary>

        /// <param name="knownContainers">Use the since variable only for the containers listed in knownContainers. Unknown containers are considered new

        /// and should be polled anyway

        /// </param>

        /// <param name="since"></param>

        /// <returns></returns>

        public IList<ObjectInfo> ListSharedObjects(HashSet<string> knownContainers, DateTimeOffset? since)

        {

            using (ThreadContext.Stacks["Share"].Push("List Objects"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                //'since' is not used here because we need to have ListObjects return a NoChange result

                //for all shared accounts,containers

                Func<ContainerInfo, string> getKey = c => String.Format("{0}\\{1}", c.Account, c.Name);

                var containers = (from account in ListSharingAccounts()

                                 let conts = TaskEx.Run(async ()=>await ListContainers(account.name).ConfigureAwait(false)).Result

                                 from container in conts

                                 select container).ToList();                

                var items = from container in containers 

                            let actualSince=knownContainers.Contains(getKey(container))?since:null

                            select ListObjects(container.Account , container.Name,  actualSince);

                var objects=items.SelectMany(r=> r).ToList();

                //For each object

                //Check parents recursively up to (but not including) the container.

                //If parents are missing, add them to the list

                //Need function to calculate all parent URLs

                objects = AddMissingParents(objects);

                //Store any new containers

                foreach (var container in containers)

                {

                    knownContainers.Add(getKey(container));

                }

                if (Log.IsDebugEnabled) Log.DebugFormat("END");

                return objects;

            }

        }

        private List<ObjectInfo> AddMissingParents(List<ObjectInfo> objects)

        {

            //TODO: Remove short-circuit when we decide to use Missing Parents functionality

            //return objects;

            var existingUris = objects.ToDictionary(o => o.Uri, o => o);

            foreach (var objectInfo in objects)

            {

                //Can be null when retrieving objects to show in selective sync

                if (objectInfo.Name == null)

                    continue;

                //No need to unescape here, the parts will be used to create new ObjectInfos

                var parts = objectInfo.Name.ToString().Split(new[]{'/'},StringSplitOptions.RemoveEmptyEntries);

                //If there is no parent, skip

                if (parts.Length == 1)

                    continue;

                var baseParts = new[]

                                  {

                                      objectInfo.Uri.Host, objectInfo.Uri.Segments[1].TrimEnd('/'),objectInfo.Account,objectInfo.Container.ToString()

                                  };

                for (var partIdx = 0; partIdx < parts.Length - 1; partIdx++)

                {

                    var nameparts = parts.Range(0, partIdx).ToArray();

                    var parentName= String.Join("/", nameparts);

                    var parentParts = baseParts.Concat(nameparts);

                    var parentUrl = objectInfo.Uri.Scheme+ "://" + String.Join("/", parentParts);

                    var parentUri = new Uri(parentUrl, UriKind.Absolute);

                    ObjectInfo existingInfo;

                    if (!existingUris.TryGetValue(parentUri,out existingInfo))

                    {

                        var h = parentUrl.GetHashCode();

                        var reverse = new string(parentUrl.Reverse().ToArray());

                        var rh = reverse.GetHashCode();

                        var b1 = BitConverter.GetBytes(h);

                        var b2 = BitConverter.GetBytes(rh);

                        var g = new Guid(0,0,0,b1.Concat(b2).ToArray());

                        existingUris[parentUri] = new ObjectInfo

                                                      {

                                                          Account = objectInfo.Account,

                                                          Container = objectInfo.Container,

                                                          Content_Type = ObjectInfo.CONTENT_TYPE_DIRECTORY,

                                                          ETag = Signature.MERKLE_EMPTY,

                                                          X_Object_Hash = Signature.MERKLE_EMPTY,

                                                          Name=new Uri(parentName,UriKind.Relative),

                                                          StorageUri=objectInfo.StorageUri,

                                                          Bytes = 0,

                                                          UUID=g.ToString(),                                                          

                                                      };

                    }

                }

            }

            return existingUris.Values.ToList();

        }

        public void SetTags(ObjectInfo target,IDictionary<string,string> tags)

        {

            Contract.Requires<InvalidOperationException>(!String.IsNullOrWhiteSpace(Token),"The Token is not set");

            Contract.Requires<InvalidOperationException>(StorageUrl != null,"The StorageUrl is not set");

            Contract.Requires<ArgumentNullException>(target != null,"target is null");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Share"].Push("Share Object"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                using (var client = new RestClient(_baseClient))

                {

                    client.BaseAddress = GetAccountUrl(target.Account);

                    client.Parameters.Clear();

                    client.Parameters.Add("update", "");

                    foreach (var tag in tags)

                    {

                        var headerTag = String.Format("X-Object-Meta-{0}", tag.Key);

                        client.Headers.Add(headerTag, tag.Value);

                    }

                    client.DownloadStringWithRetryRelative(target.Container, 3);

                    client.AssertStatusOK("SetTags failed");

                    //If the status is NOT ACCEPTED we have a problem

                    if (client.StatusCode != HttpStatusCode.Accepted)

                    {

                        Log.Error("Failed to set tags");

                        throw new Exception("Failed to set tags");

                    }

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                }

            }

        }

        public void ShareObject(string account, Uri container, Uri objectName, string shareTo, bool read, bool write)

        {

            Contract.Requires<InvalidOperationException>(!String.IsNullOrWhiteSpace(Token), "The Token is not set");

            Contract.Requires<InvalidOperationException>(StorageUrl != null, "The StorageUrl is not set");

            Contract.Requires<ArgumentNullException>(container != null, "container is null");

            Contract.Requires<ArgumentException>(!container.IsAbsoluteUri, "container is absolute");

            Contract.Requires<ArgumentNullException>(objectName != null, "objectName is null");

            Contract.Requires<ArgumentException>(!objectName.IsAbsoluteUri, "objectName  is absolute");

            Contract.Requires<ArgumentNullException>(!String.IsNullOrWhiteSpace(account), "account is not set");

            Contract.Requires<ArgumentNullException>(!String.IsNullOrWhiteSpace(shareTo), "shareTo is not set");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Share"].Push("Share Object"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                using (var client = new RestClient(_baseClient))

                {

                    client.BaseAddress = GetAccountUrl(account);

                    client.Parameters.Clear();

                    client.Parameters.Add("format", "json");

                    string permission = "";

                    if (write)

                        permission = String.Format("write={0}", shareTo);

                    else if (read)

                        permission = String.Format("read={0}", shareTo);

                    client.Headers.Add("X-Object-Sharing", permission);

                    var content = client.DownloadStringWithRetryRelative(container, 3);

                    client.AssertStatusOK("ShareObject failed");

                    //If the result is empty, return an empty list,

                    var infos = String.IsNullOrWhiteSpace(content)

                                    ? new List<ObjectInfo>()

                                //Otherwise deserialize the object list into a list of ObjectInfos

                                    : JsonConvert.DeserializeObject<IList<ObjectInfo>>(content);

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                }

            }

        }

        public async Task<AccountInfo> GetAccountPolicies(AccountInfo accountInfo)

        {

            if (accountInfo==null)

                throw new ArgumentNullException("accountInfo");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Account"].Push("GetPolicies"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

/*

                if (_baseClient == null)

                {

                    _baseClient = new RestClient

                    {

                        BaseAddress = accountInfo.StorageUri.ToString(),

                        Timeout = 30000,

                        Retries = 3,

                    };

                }

*/                

                var containerUri = GetTargetUri(accountInfo.UserName);

                var targetUri = new Uri(String.Format("{0}?format=json", containerUri), UriKind.Absolute);

                using(var response=await _baseHttpClient.HeadAsyncWithRetries(targetUri,3).ConfigureAwait(false))

                {

                    var quotaValue=response.Headers.GetFirstValue("X-Account-Policy-Quota");

                    var bytesValue = response.Headers.GetFirstValue("X-Account-Bytes-Used");

                    long quota, bytes;

                    if (long.TryParse(quotaValue, out quota))

                        accountInfo.Quota = quota;

                    if (long.TryParse(bytesValue, out bytes))

                        accountInfo.BytesUsed = bytes;

                    return accountInfo;   

                }

                //using (var client = new RestClient(_baseClient))

                //{

                //    if (!String.IsNullOrWhiteSpace(accountInfo.UserName))

                //        client.BaseAddress = GetAccountUrl(accountInfo.UserName);

                //    client.Parameters.Clear();

                //    client.Parameters.Add("format", "json");                    

                //    client.Head(_emptyUri, 3);

                //    var quotaValue=client.ResponseHeaders["X-Account-Policy-Quota"];

                //    var bytesValue= client.ResponseHeaders["X-Account-Bytes-Used"];

                //    long quota, bytes;

                //    if (long.TryParse(quotaValue, out quota))

                //        accountInfo.Quota = quota;

                //    if (long.TryParse(bytesValue, out bytes))

                //        accountInfo.BytesUsed = bytes;

                //    return accountInfo;

                //}

            }

        }

        public void UpdateMetadata(ObjectInfo objectInfo)

        {

            Contract.Requires<ArgumentNullException>(objectInfo != null,"objectInfo is null");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Objects"].Push("UpdateMetadata"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                using(var client=new RestClient(_baseClient))

                {

                    client.BaseAddress = GetAccountUrl(objectInfo.Account);

                    client.Parameters.Clear();

                    //Set Tags

                    foreach (var tag in objectInfo.Tags)

                    {

                        var headerTag = String.Format("X-Object-Meta-{0}", tag.Key);

                        client.Headers.Add(headerTag, tag.Value);

                    }

                    //Set Permissions

                    var permissions=objectInfo.GetPermissionString();

                    client.SetNonEmptyHeaderValue("X-Object-Sharing",permissions);

                    client.SetNonEmptyHeaderValue("Content-Disposition",objectInfo.ContendDisposition);

                    client.SetNonEmptyHeaderValue("Content-Encoding",objectInfo.ContentEncoding);

                    client.SetNonEmptyHeaderValue("X-Object-Manifest",objectInfo.Manifest);

                    var isPublic = objectInfo.IsPublic.ToString().ToLower();

                    client.Headers.Add("X-Object-Public", isPublic);

                    var address = String.Format("{0}/{1}?update=",objectInfo.Container, objectInfo.Name);

                    client.PostWithRetry(new Uri(address,UriKind.Relative),"application/xml");

                    client.AssertStatusOK("UpdateMetadata failed");

                    //If the status is NOT ACCEPTED or OK we have a problem

                    if (!(client.StatusCode == HttpStatusCode.Accepted || client.StatusCode == HttpStatusCode.OK))

                    {

                        Log.Error("Failed to update metadata");

                        throw new Exception("Failed to update metadata");

                    }

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                }

            }

        }

        public void UpdateMetadata(ContainerInfo containerInfo)

        {

            if (containerInfo == null)

                throw new ArgumentNullException("containerInfo");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Containers"].Push("UpdateMetadata"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                using(var client=new RestClient(_baseClient))

                {

                    client.BaseAddress = GetAccountUrl(containerInfo.Account);

                    client.Parameters.Clear();

                    //Set Tags

                    foreach (var tag in containerInfo.Tags)

                    {

                        var headerTag = String.Format("X-Container-Meta-{0}", tag.Key);

                        client.Headers.Add(headerTag, tag.Value);

                    }

                    //Set Policies

                    foreach (var policy in containerInfo.Policies)

                    {

                        var headerPolicy = String.Format("X-Container-Policy-{0}", policy.Key);

                        client.Headers.Add(headerPolicy, policy.Value);

                    }

                    var uriBuilder = client.GetAddressBuilder(containerInfo.Name,_emptyUri);

                    var uri = uriBuilder.Uri;

                    client.UploadValues(uri,new NameValueCollection());

                    client.AssertStatusOK("UpdateMetadata failed");

                    //If the status is NOT ACCEPTED or OK we have a problem

                    if (!(client.StatusCode == HttpStatusCode.Accepted || client.StatusCode == HttpStatusCode.OK))

                    {

                        Log.Error("Failed to update metadata");

                        throw new Exception("Failed to update metadata");

                    }

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                }

            }

        }

        public IList<ObjectInfo> ListObjects(string account, Uri container, DateTimeOffset? since = null)

        {

/*

            if (container==null)

                throw new ArgumentNullException("container");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("container");

            Contract.EndContractBlock();

*/

            using (ThreadContext.Stacks["Objects"].Push("List"))

            {

                var containerUri = GetTargetUri(account).Combine(container);

                var targetUri = new Uri(String.Format("{0}?format=json", containerUri), UriKind.Absolute);

                var content =TaskEx.Run(async ()=>await GetStringAsync(targetUri, "ListObjects failed", since).ConfigureAwait(false)).Result;

                //304 will result in an empty string. Empty containers return an empty json array

                if (String.IsNullOrWhiteSpace(content))

                     return new[] {new NoModificationInfo(account, container)};

                 //If the result is empty, return an empty list,

                 var infos = String.IsNullOrWhiteSpace(content)

                                 ? new List<ObjectInfo>()

                             //Otherwise deserialize the object list into a list of ObjectInfos

                                 : JsonConvert.DeserializeObject<IList<ObjectInfo>>(content);

                 foreach (var info in infos)

                 {

                     info.Container = container;

                     info.Account = account;

                     info.StorageUri = StorageUrl;

                 }

                 if (Log.IsDebugEnabled) Log.DebugFormat("END");

                 return infos;

            }

        }

        public IList<ObjectInfo> ListObjects(string account, Uri container, Uri folder, DateTimeOffset? since = null)

        {

/*            if (container==null)

                throw new ArgumentNullException("container");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("container");

            Contract.EndContractBlock();*/

            using (ThreadContext.Stacks["Objects"].Push("List"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                var containerUri = GetTargetUri(account).Combine(container);

                var targetUri = new Uri(String.Format("{0}?format=json&path={1}", containerUri,folder), UriKind.Absolute);

                var content = TaskEx.Run(async ()=>await GetStringAsync(targetUri, "ListObjects failed", since).ConfigureAwait(false)).Result;                

                //304 will result in an empty string. Empty containers return an empty json array

                if (String.IsNullOrWhiteSpace(content))

                    return new[] { new NoModificationInfo(account, container) };

                var infos = JsonConvert.DeserializeObject<IList<ObjectInfo>>(content);

                foreach (var info in infos)

                {

                    info.Account = account;

                    if (info.Container == null)

                        info.Container = container;

                    info.StorageUri = StorageUrl;

                }

                if (Log.IsDebugEnabled) Log.DebugFormat("END");

                return infos;

/*

                using (var client = new RestClient(_baseClient))

                {

                    if (!String.IsNullOrWhiteSpace(account))

                        client.BaseAddress = GetAccountUrl(account);

                    client.Parameters.Clear();

                    client.Parameters.Add("format", "json");

                    client.Parameters.Add("path", folder.ToString());

                    client.IfModifiedSince = since;

                    var content = client.DownloadStringWithRetryRelative(container, 3);

                    client.AssertStatusOK("ListObjects failed");

                    if (client.StatusCode==HttpStatusCode.NotModified)

                        return new[]{new NoModificationInfo(account,container,folder)};

                    var infos = JsonConvert.DeserializeObject<IList<ObjectInfo>>(content);

                    foreach (var info in infos)

                    {

                        info.Account = account;

                        if (info.Container == null)

                            info.Container = container;

                        info.StorageUri = StorageUrl;

                    }

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                    return infos;

                }

*/

            }

        }

        public async Task<bool> ContainerExists(string account, Uri container)

        {

            if (container==null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException( "The container must be relative","container");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Containters"].Push("Exists"))

            {

                if (Log.IsDebugEnabled) Log.DebugFormat("START");

                var targetUri = GetTargetUri(account).Combine(container);

                using (var response =await _baseHttpClient.HeadAsyncWithRetries(targetUri, 3))

                {

                    bool result;

                    switch (response.StatusCode)

                    {

                        case HttpStatusCode.OK:

                        case HttpStatusCode.NoContent:

                            result = true;

                            break;

                        case HttpStatusCode.NotFound:

                            result = false;

                            break;

                        default:

                            throw CreateWebException("ContainerExists", response.StatusCode);

                    }

                    if (Log.IsDebugEnabled) Log.DebugFormat("END");

                    return result;

                }

            }

        }

        private Uri GetTargetUri(string account)

        {

            return new Uri(GetTargetUrl(account),UriKind.Absolute);

        }

        private string GetTargetUrl(string account)

        {

            return String.IsNullOrWhiteSpace(account)

                       ? _baseHttpClient.BaseAddress.ToString()

                       : GetAccountUrl(account);

        }

        public async Task<bool> ObjectExists(string account, Uri container, Uri objectName)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (objectName == null)

                throw new ArgumentNullException("objectName", "The objectName property can't be empty");

            if (objectName.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative","objectName");

            Contract.EndContractBlock();

                var targetUri=GetTargetUri(account).Combine(container).Combine(objectName);

            using (var response = await _baseHttpClient.HeadAsyncWithRetries(targetUri, 3).ConfigureAwait(false))

            {

                switch (response.StatusCode)

                {

                    case HttpStatusCode.OK:

                    case HttpStatusCode.NoContent:

                        return true;

                    case HttpStatusCode.NotFound:

                        return false;

                    default:

                        throw CreateWebException("ObjectExists", response.StatusCode);

                }

            }

        }

        public async Task<ObjectInfo> GetObjectInfo(string account, Uri container, Uri objectName)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative", "container");

            if (objectName == null)

                throw new ArgumentNullException("objectName", "The objectName property can't be empty");

            if (objectName.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative", "objectName");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Objects"].Push("GetObjectInfo"))

            {

                var targetUri = GetTargetUri(account).Combine(container).Combine(objectName);

                try

                {

                    using (var response = await _baseHttpClient.HeadAsyncWithRetries(targetUri, 3,true))

                    {

                        switch (response.StatusCode)

                        {

                            case HttpStatusCode.OK:

                            case HttpStatusCode.NoContent:

                                var tags = response.Headers.GetMeta("X-Object-Meta-");

                                var extensions = (from header in response.Headers

                                                  where

                                                      header.Key.StartsWith("X-Object-") &&

                                                      !header.Key.StartsWith("X-Object-Meta-")

                                                  select new {Name = header.Key, Value = header.Value.FirstOrDefault()})

                                    .ToDictionary(t => t.Name, t => t.Value);

                                var permissions = response.Headers.GetFirstValue("X-Object-Sharing");

                                var info = new ObjectInfo

                                               {

                                                   Account = account,

                                                   Container = container,

                                                   Name = objectName,

                                                   ETag = response.Headers.ETag.NullSafe(e=>e.Tag),

                                                   UUID = response.Headers.GetFirstValue("X-Object-UUID"),

                                                   X_Object_Hash = response.Headers.GetFirstValue("X-Object-Hash"),

                                                   Content_Type = response.Headers.GetFirstValue("Content-Type"),

                                                   Bytes = Convert.ToInt64(response.Content.Headers.ContentLength),

                                                   Tags = tags,

                                                   Last_Modified = response.Content.Headers.LastModified,

                                                   Extensions = extensions,

                                                   ContentEncoding =

                                                       response.Content.Headers.ContentEncoding.FirstOrDefault(),

                                                   ContendDisposition =

                                                       response.Content.Headers.ContentDisposition.NullSafe(c=>c.ToString()),

                                                   Manifest = response.Headers.GetFirstValue("X-Object-Manifest"),

                                                   PublicUrl = response.Headers.GetFirstValue("X-Object-Public"),

                                                   StorageUri = StorageUrl,

                                               };

                                info.SetPermissions(permissions);

                                return info;

                            case HttpStatusCode.NotFound:

                                return ObjectInfo.Empty;

                            default:

                                throw new WebException(

                                    String.Format("[FAIL] GetObjectInfo for {0} failed with unexpected status code {1}",

                                                  objectName, response.StatusCode));

                        }

                    }

                }

                catch (RetryException)

                {

                    Log.WarnFormat("[RETRY FAIL] GetObjectInfo for {0} failed.", objectName);

                    return ObjectInfo.Empty;

                }

                catch (WebException e)

                {

                    Log.Error(

                        String.Format("[FAIL] GetObjectInfo for {0} failed with unexpected status {1}",

                                      objectName, e.Status), e);

                    throw;

                }

            }

        }

        public async Task CreateFolder(string account, Uri container, Uri folder)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (folder == null)

                throw new ArgumentNullException("folder", "The objectName property can't be empty");

            if (folder.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative","folder");

            Contract.EndContractBlock();

            var folderUri=container.Combine(folder);            

            var targetUri = GetTargetUri(account).Combine(folderUri);

            var message = new HttpRequestMessage(HttpMethod.Put, targetUri);

            message.Content=new StringContent("");

            message.Content.Headers.ContentType = new MediaTypeHeaderValue(ObjectInfo.CONTENT_TYPE_DIRECTORY);

            //message.Headers.Add("Content-Length", "0");

            using (var response = await _baseHttpClient.SendAsyncWithRetries(message, 3).ConfigureAwait(false))

            {

                if (response.StatusCode != HttpStatusCode.Created && response.StatusCode != HttpStatusCode.Accepted)

                    throw CreateWebException("CreateFolder", response.StatusCode);

            }

        }

        private Dictionary<string, string> GetMeta(HttpResponseMessage response,string metaPrefix)

        {

            Contract.Requires<ArgumentNullException>(!String.IsNullOrWhiteSpace(metaPrefix), "metaPrefix is empty");

            Contract.EndContractBlock();

            var dict = (from header in response.Headers

                        where header.Key.StartsWith(metaPrefix)

                         select new { Name = header.Key, Value = String.Join(",", header.Value) })

                        .ToDictionary(t => t.Name, t => t.Value);

            return dict;

        }

        public async Task<ContainerInfo> GetContainerInfo(string account, Uri container)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            Contract.EndContractBlock();

            var targetUri = GetTargetUri(account).Combine(container);            

            using (var response = await _baseHttpClient.HeadAsyncWithRetries(targetUri, 3,true).ConfigureAwait(false))

            {

                if (Log.IsDebugEnabled)

                    Log.DebugFormat("ContainerInfo data: {0}\n{1}",response,await response.Content.ReadAsStringAsync().ConfigureAwait(false));

                switch (response.StatusCode)

                {

                    case HttpStatusCode.OK:

                    case HttpStatusCode.NoContent:

                        var tags = GetMeta(response,"X-Container-Meta-");

                        var policies = GetMeta(response,"X-Container-Policy-");

                        var containerInfo = new ContainerInfo

                                                {

                                                    Account = account,

                                                    Name = container,

                                                    StorageUrl = StorageUrl.ToString(),

                                                    Count =long.Parse(response.Headers.GetFirstValue("X-Container-Object-Count")),

                                                    Bytes = long.Parse(response.Headers.GetFirstValue("X-Container-Bytes-Used")),

                                                    BlockHash = response.Headers.GetFirstValue("X-Container-Block-Hash"),

                                                    BlockSize =

                                                        int.Parse(response.Headers.GetFirstValue("X-Container-Block-Size")),

                                                    Last_Modified = response.Content.Headers.LastModified,

                                                    Tags = tags,

                                                    Policies = policies

                                                };

                        return containerInfo;

                    case HttpStatusCode.NotFound:

                        return ContainerInfo.Empty;

                    default:

                        throw CreateWebException("GetContainerInfo", response.StatusCode);

                }

            }            

        }

        public async Task CreateContainer(string account, Uri container)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            Contract.EndContractBlock();

            var targetUri=GetTargetUri(account).Combine(container);

            var message = new HttpRequestMessage(HttpMethod.Put, targetUri);

            //message.Content.Headers.ContentLength = 0;

            using (var response =await _baseHttpClient.SendAsyncWithRetries(message, 3).ConfigureAwait(false))

            {            

                var expectedCodes = new[] {HttpStatusCode.Created, HttpStatusCode.Accepted, HttpStatusCode.OK};

                if (!expectedCodes.Contains(response.StatusCode))

                    throw CreateWebException("CreateContainer", response.StatusCode);

            }

        }

        public async Task WipeContainer(string account, Uri container)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative", "container");

            Contract.EndContractBlock();

            await DeleteContainer(account, new Uri(String.Format("{0}?delimiter=/", container), UriKind.Relative)).ConfigureAwait(false);

        }

        public async Task DeleteContainer(string account, Uri container)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            Contract.EndContractBlock();

            var targetUri = GetTargetUri(account).Combine(container);

            var message = new HttpRequestMessage(HttpMethod.Delete, targetUri);

            using (var response = await _baseHttpClient.SendAsyncWithRetries(message, 3).ConfigureAwait(false))

            {

                var expectedCodes = new[] { HttpStatusCode.NotFound, HttpStatusCode.NoContent };

                if (!expectedCodes.Contains(response.StatusCode))

                    throw CreateWebException("DeleteContainer", response.StatusCode);

            }

        }

        /// <summary>

        /// 

        /// </summary>

        /// <param name="account"></param>

        /// <param name="container"></param>

        /// <param name="objectName"></param>

        /// <param name="fileName"></param>

        /// <param name="cancellationToken"> </param>

        /// <returns></returns>

        /// <remarks>This method should have no timeout or a very long one</remarks>

        //Asynchronously download the object specified by *objectName* in a specific *container* to 

        // a local file

        public async Task GetObject(string account, Uri container, Uri objectName, string fileName,CancellationToken cancellationToken)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (objectName == null)

                throw new ArgumentNullException("objectName", "The objectName property can't be empty");

            if (objectName.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative","objectName");

            Contract.EndContractBlock();

            try

            {

                //WebClient, and by extension RestClient, are not thread-safe. Create a new RestClient

                //object to avoid concurrency errors.

                //

                //Download operations take a long time therefore they have no timeout.

                using(var client = new RestClient(_baseClient) { Timeout = 0 })

                {

                    if (!String.IsNullOrWhiteSpace(account))

                        client.BaseAddress = GetAccountUrl(account);

                    //The container and objectName are relative names. They are joined with the client's

                    //BaseAddress to create the object's absolute address

                    var builder = client.GetAddressBuilder(container, objectName);

                    var uri = builder.Uri;

                    //Download progress is reported to the Trace log

                    Log.InfoFormat("[GET] START {0}", objectName);

                    /*client.DownloadProgressChanged += (sender, args) =>

                                                      Log.InfoFormat("[GET PROGRESS] {0} {1}% {2} of {3}",

                                                                     fileName, args.ProgressPercentage,

                                                                     args.BytesReceived,

                                                                     args.TotalBytesToReceive);*/

                    var progress = new Progress<DownloadProgressChangedEventArgs>(args =>

                                {

                                    Log.InfoFormat("[GET PROGRESS] {0} {1}% {2} of {3}",

                                                   fileName, args.ProgressPercentage,

                                                   args.BytesReceived,

                                                   args.TotalBytesToReceive);

                                    if (DownloadProgressChanged!=null)

                                        DownloadProgressChanged(this, new DownloadArgs(args));

                                });

                    //Start downloading the object asynchronously                    

                    await client.DownloadFileTaskAsync(uri, fileName, cancellationToken,progress).ConfigureAwait(false);

                    //Once the download completes

                    //Delete the local client object

                }

                //And report failure or completion

            }

            catch (Exception exc)

            {

                Log.ErrorFormat("[GET] FAIL {0} with {1}", objectName, exc);

                throw;

            }

            Log.InfoFormat("[GET] END {0}", objectName);                                             

        }

        public async Task<IList<string>> PutHashMap(string account, Uri container, Uri objectName, TreeHash hash)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (objectName == null)

                throw new ArgumentNullException("objectName", "The objectName property can't be empty");

            if (objectName.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative","objectName");

            if (hash == null)

                throw new ArgumentNullException("hash");

            if (String.IsNullOrWhiteSpace(Token))

                throw new InvalidOperationException("Invalid Token");

            if (StorageUrl == null)

                throw new InvalidOperationException("Invalid Storage Url");

            Contract.EndContractBlock();

            //The container and objectName are relative names. They are joined with the client's

            //BaseAddress to create the object's absolute address

            var targetUri = GetTargetUri(account).Combine(container).Combine(objectName);

            var uri = new Uri(String.Format("{0}?format=json&hashmap",targetUri),UriKind.Absolute);

            //Send the tree hash as Json to the server            

            var jsonHash = hash.ToJson();

            if (Log.IsDebugEnabled)

                Log.DebugFormat("Hashes:\r\n{0}", jsonHash);

            var mimeType = objectName.GetMimeType();

            var message = new HttpRequestMessage(HttpMethod.Put, uri)

            {

                Content = new StringContent(jsonHash)

            };

            message.Content.Headers.ContentType = mimeType;

            message.Headers.Add("ETag",hash.TopHash.ToHashString());

            //Don't use a timeout because putting the hashmap may be a long process

            using (var response = await _baseHttpClientNoTimeout.SendAsyncWithRetries(message, 3).ConfigureAwait(false))

            {

                var empty = (IList<string>)new List<string>();

                switch (response.StatusCode)

                {

                    case HttpStatusCode.Created:

                        //The server will respond either with 201-created if all blocks were already on the server

                        return empty;

                    case HttpStatusCode.Conflict:

                        //or with a 409-conflict and return the list of missing parts

                        using (var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false))

                        using(var reader=stream.GetLoggedReader(Log))

                        {                            

                            var serializer = new JsonSerializer();                            

                            serializer.Error += (sender, args) => Log.ErrorFormat("Deserialization error at [{0}] [{1}]", args.ErrorContext.Error, args.ErrorContext.Member);

                            var hashes = (List<string>)serializer.Deserialize(reader, typeof(List<string>));

                            return hashes;

                        }                        

                    default:

                        //All other cases are unexpected

                        //Ensure that failure codes raise exceptions

                        response.EnsureSuccessStatusCode();

                        //And log any other codes as warngings, but continute processing

                        Log.WarnFormat("Unexcpected status code when putting map: {0} - {1}",response.StatusCode,response.ReasonPhrase);

                        return empty;

                }

            }

        }

        public async Task<byte[]> GetBlock(string account, Uri container, Uri relativeUrl, long start, long? end, CancellationToken cancellationToken)

        {

            if (String.IsNullOrWhiteSpace(Token))

                throw new InvalidOperationException("Invalid Token");

            if (StorageUrl == null)

                throw new InvalidOperationException("Invalid Storage Url");

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (relativeUrl == null)

                throw new ArgumentNullException("relativeUrl");

            if (end.HasValue && end < 0)

                throw new ArgumentOutOfRangeException("end");

            if (start < 0)

                throw new ArgumentOutOfRangeException("start");

            Contract.EndContractBlock();

            var targetUri = GetTargetUri(account).Combine(container).Combine(relativeUrl);

            var message = new HttpRequestMessage(HttpMethod.Get, targetUri);

            //Don't add a range if start=0, end=null (empty files)

            if (start!=0 || end!=null)

                message.Headers.Range=new RangeHeaderValue(start,end);

            //Don't use a timeout because putting the hashmap may be a long process

            IProgress<DownloadArgs> progress = new Progress<DownloadArgs>(args =>

                {

                    Log.DebugFormat("[GET PROGRESS] {0} {1}% {2} of {3}",

                                    targetUri.Segments.Last(), args.ProgressPercentage,

                                    args.BytesReceived,

                                    args.TotalBytesToReceive);

                    if (DownloadProgressChanged!=null)

                        DownloadProgressChanged(this,  args);

                });

            using (var response = await _baseHttpClientNoTimeout.SendAsyncWithRetries(message, 3, false,HttpCompletionOption.ResponseHeadersRead,

                                                          cancellationToken).ConfigureAwait(false))

            using (var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false))

            {

                long totalSize = response.Content.Headers.ContentLength ?? 0;

                    byte[] buffer,streambuf;

                    lock (_bufferManager)

                    {

                        buffer = _bufferManager.TakeBuffer(65536);

                        streambuf = _bufferManager.TakeBuffer((int)totalSize);

                    }

                using (var targetStream = new MemoryStream(streambuf))

                {

                    long total = 0;

                    try

                    {

                        int read;

                        while ((read = await stream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false)) != 0)

                        {

                            total += read;

                            progress.Report(new DownloadArgs(total, totalSize));

                            await targetStream.WriteAsync(buffer, 0, read).ConfigureAwait(false);

                        }

                    }

                    finally

                    {

                        lock (_bufferManager)

                        {

                            _bufferManager.ReturnBuffer(buffer);

                            _bufferManager.ReturnBuffer(streambuf);

                        }

                    }

                    var result = targetStream.ToArray();

                    return result;

                }

            }

        }

        public event EventHandler<UploadArgs> UploadProgressChanged;

        public event EventHandler<DownloadArgs> DownloadProgressChanged;

        public async Task PostBlock(string account, Uri container, byte[] block, int offset, int count,string blockHash,CancellationToken token)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (block == null)

                throw new ArgumentNullException("block");

            if (offset < 0 || offset >= block.Length)

                throw new ArgumentOutOfRangeException("offset");

            if (count < 0 || count > block.Length)

                throw new ArgumentOutOfRangeException("count");

            if (String.IsNullOrWhiteSpace(Token))

                throw new InvalidOperationException("Invalid Token");

            if (StorageUrl == null)

                throw new InvalidOperationException("Invalid Storage Url");                        

            Contract.EndContractBlock();

            try

            {

                var containerUri = GetTargetUri(account).Combine(container);

                var targetUri = new Uri(String.Format("{0}?update", containerUri));

                //Don't use a timeout because putting the hashmap may be a long process

                Log.InfoFormat("[BLOCK POST] START");

                var progress = new Progress<UploadArgs>(args =>

                {

                    Log.InfoFormat("[BLOCK POST PROGRESS] {0}% {1} of {2}",

                        args.ProgressPercentage,

                        args.BytesSent,

                        args.TotalBytesToSend);

                    if (UploadProgressChanged != null)

                        UploadProgressChanged(this,args);

                });

                var message = new HttpRequestMessage(HttpMethod.Post, targetUri)

                                  {

                                      Content = new ByteArrayContentWithProgress(block, offset, count,progress)

                                  };

                message.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(@"application/octet-stream");

                //Send the block

                using (var response = await _baseHttpClientNoTimeout.SendAsyncWithRetries(message, 3,false,HttpCompletionOption.ResponseContentRead,token).ConfigureAwait(false))

                {                    

                    Log.InfoFormat("[BLOCK POST PROGRESS] Completed ");

                    response.EnsureSuccessStatusCode();

                    var responseHash = await response.Content.ReadAsStringAsync().ConfigureAwait(false);

                    var cleanHash = responseHash.TrimEnd();

                    Debug.Assert(blockHash==cleanHash);

                    if (!blockHash.Equals(cleanHash, StringComparison.OrdinalIgnoreCase))

                        Log.ErrorFormat("Block hash mismatch posting to [{0}]:[{1}], expected [{2}] but was [{3}]", account, container, blockHash, cleanHash);

                }

                Log.InfoFormat("[BLOCK POST] END");               

            }

            catch (TaskCanceledException )

            {

                Log.Info("Aborting block");

                throw;

            }

            catch (Exception exc)

            {

                Log.ErrorFormat("[BLOCK POST] FAIL with \r{0}", exc);

                throw;

            }

        }

        public async Task PostBlock(string account, Uri container, string filePath, long offset, int count, string blockHash, CancellationToken token)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative", "container");

            if (String.IsNullOrWhiteSpace(filePath))

                throw new ArgumentNullException("filePath");

            if (!File.Exists(filePath))

                throw new FileNotFoundException("Missing file","filePath");

            if (String.IsNullOrWhiteSpace(Token))

                throw new InvalidOperationException("Invalid Token");

            if (StorageUrl == null)

                throw new InvalidOperationException("Invalid Storage Url");

            Contract.EndContractBlock();

            try

            {

                var containerUri = GetTargetUri(account).Combine(container);

                var targetUri = new Uri(String.Format("{0}?update", containerUri));

                //Don't use a timeout because putting the hashmap may be a long process

                Log.InfoFormat("[BLOCK POST] START");

                var progress = new Progress<UploadArgs>(args =>

                {

                    Log.InfoFormat("[BLOCK POST PROGRESS] {0}% {1} of {2} at {3:###,} Kbps ",

                        args.ProgressPercentage,

                        args.BytesSent,

                        args.TotalBytesToSend,args.Speed);

                    if (UploadProgressChanged != null)

                        UploadProgressChanged(this, args);

                });

                var message = new HttpRequestMessage(HttpMethod.Post, targetUri)

                {

                    Content = new FileBlockContent(filePath, offset, count, progress)

                };

                message.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(@"application/octet-stream");

                //Send the block

                using (var response = await _baseHttpClientNoTimeout.SendAsyncWithRetries(message, 3, false, HttpCompletionOption.ResponseContentRead, token).ConfigureAwait(false))

                {

                    Log.InfoFormat("[BLOCK POST PROGRESS] Completed ");

                    response.EnsureSuccessStatusCode();

                    var responseHash = await response.Content.ReadAsStringAsync().ConfigureAwait(false);

                    var cleanHash = responseHash.TrimEnd();

                    Debug.Assert(blockHash == cleanHash);

                    if (!blockHash.Equals(cleanHash, StringComparison.OrdinalIgnoreCase))

                        Log.ErrorFormat("Block hash mismatch posting to [{0}]:[{1}], expected [{2}] but was [{3}]", account, container, blockHash, cleanHash);

                }

                Log.InfoFormat("[BLOCK POST] END");

            }

            catch (TaskCanceledException)

            {

                Log.Info("Aborting block");

                throw;

            }                

            catch (Exception exc)

            {

                Log.ErrorFormat("[BLOCK POST] FAIL with \r{0}", exc);

                throw;

            }

        }

        public async Task<TreeHash> GetHashMap(string account, Uri container, Uri objectName)

        {

            if (container == null)

                throw new ArgumentNullException("container", "The container property can't be empty");

            if (container.IsAbsoluteUri)

                throw new ArgumentException("The container must be relative","container");

            if (objectName == null)

                throw new ArgumentNullException("objectName", "The objectName property can't be empty");

            if (objectName.IsAbsoluteUri)

                throw new ArgumentException("The objectName must be relative","objectName");

            if (String.IsNullOrWhiteSpace(Token))

                throw new InvalidOperationException("Invalid Token");

            if (StorageUrl == null)

                throw new InvalidOperationException("Invalid Storage Url");

            Contract.EndContractBlock();

            try

            {

                var objectUri = GetTargetUri(account).Combine(container).Combine(objectName);

                var targetUri = new Uri(String.Format("{0}?format=json&hashmap", objectUri));

                //Start downloading the object asynchronously

                var json = await GetStringAsync(targetUri, "").ConfigureAwait(false);

                var treeHash = TreeHash.Parse(json);

                Log.InfoFormat("[GET HASH] END {0}", objectName);

                return treeHash;

            }

            catch (Exception exc)

            {

                Log.ErrorFormat("[GET HASH] END {0} with {1}", objectName, exc);

                throw;

            }

        }

        /// <summary>

        /// 

        /// </summary>

        /// <param name="account"></param>

        /// <param name="container"></param>

        /// <param name="objectName"></param>