Synnefo » Pithos MS Client

#region

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

 * <copyright file="RestClient.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

using System.Collections.Specialized;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Net;

using System.Reflection;

using System.Runtime.Serialization;

using System.Threading;

using System.Threading.Tasks;

using log4net;

namespace Pithos.Network

{

    using System;

    using System.Collections.Generic;

    using System.Linq;

    using System.Text;

    /// <summary>

    /// TODO: Update summary.

    /// </summary>

    public class RestClient:WebClient

    {

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

        public int Timeout { get; set; }

        public bool TimedOut { get; set; }

        public HttpStatusCode StatusCode { get; private set; }

        public string StatusDescription { get; set; }

        public long? RangeFrom { get; set; }

        public long? RangeTo { get; set; }

        public int Retries { get; set; }

        private readonly Dictionary<string, string> _parameters=new Dictionary<string, string>();

        public Dictionary<string, string> Parameters

        {

            get

            {

                Contract.Ensures(_parameters!=null);

                return _parameters;

            }            

        }

        [ContractInvariantMethod]

        private void Invariants()

        {

            Contract.Invariant(Headers!=null);    

        }

        public RestClient():base()

        {

            //The maximum error response must be large because missing server hashes are return as a Conflivt (409) error response

            //Any value above 2^21-1 will result in an empty response.

            //-1 essentially ignores the maximum length

            HttpWebRequest.DefaultMaximumErrorResponseLength = -1;

        }

        public RestClient(RestClient other)

            : base()

        {

            if (other==null)

                //Log.ErrorFormat("[ERROR] No parameters provided to the rest client. \n{0}\n", other);

                throw new ArgumentNullException("other");

            Contract.EndContractBlock();

            //The maximum error response must be large because missing server hashes are return as a Conflivt (409) error response

            //Any value above 2^21-1 will result in an empty response.

            //-1 essentially ignores the maximum length

            HttpWebRequest.DefaultMaximumErrorResponseLength = -1;

            CopyHeaders(other);

            Timeout = other.Timeout;

            Retries = other.Retries;

            BaseAddress = other.BaseAddress;             

            foreach (var parameter in other.Parameters)

            {

                Parameters.Add(parameter.Key,parameter.Value);

            }

            this.Proxy = other.Proxy;

        }

        protected override WebRequest GetWebRequest(Uri address)

        {

            TimedOut = false;

            var webRequest = base.GetWebRequest(address);            

            var request = (HttpWebRequest)webRequest;

            request.CookieContainer=new CookieContainer();

            request.ServicePoint.ConnectionLimit = 50;

            if (IfModifiedSince.HasValue)

                request.IfModifiedSince = IfModifiedSince.Value;

            request.AutomaticDecompression = DecompressionMethods.Deflate | DecompressionMethods.GZip;

            if(Timeout>0)

                request.Timeout = Timeout;

            if (RangeFrom.HasValue)

            {

                if (RangeTo.HasValue)

                    request.AddRange(RangeFrom.Value, RangeTo.Value);

                else

                    request.AddRange(RangeFrom.Value);

            }

            return request; 

        }

        public DateTime? IfModifiedSince { get; set; }

        //Asynchronous version

        protected override WebResponse GetWebResponse(WebRequest request, IAsyncResult result)

        {            

            Log.InfoFormat("[{0}] {1}", request.Method, request.RequestUri); 

            HttpWebResponse response = null;

            try

            {

                response = (HttpWebResponse)base.GetWebResponse(request, result);

            }

            catch (WebException exc)

            {

                if (!TryGetResponse(exc, request,out response))

                    throw;

            }

            StatusCode = response.StatusCode;

            LastModified = response.LastModified;

            StatusDescription = response.StatusDescription;

            return response;

        }

        //Synchronous version

        protected override WebResponse GetWebResponse(WebRequest request)

        {

            HttpWebResponse response = null;

            try

            {           

                Log.InfoFormat("[{0}] {1}",request.Method,request.RequestUri);     

                response = (HttpWebResponse)base.GetWebResponse(request);

            }

            catch (WebException exc)

            {

                if (!TryGetResponse(exc, request,out response))

                    throw;

            }

            StatusCode = response.StatusCode;

            LastModified = response.LastModified;

            StatusDescription = response.StatusDescription;

            return response;

        }

        private bool TryGetResponse(WebException exc, WebRequest request,out HttpWebResponse response)

        {

            response = null;

            //Fail on empty response

            if (exc.Response == null)

            {

                Log.WarnFormat("[{0}] {1} {2}", request.Method, exc.Status, request.RequestUri);     

                return false;

            }

            response = (exc.Response as HttpWebResponse);

            var statusCode = (int)response.StatusCode;

            //Succeed on allowed status codes

            if (AllowedStatusCodes.Contains(response.StatusCode))

            {

                if (Log.IsDebugEnabled)

                    Log.DebugFormat("[{0}] {1} {2}", request.Method, statusCode, request.RequestUri);     

                return true;

            }

            Log.WarnFormat("[{0}] {1} {2}", request.Method, statusCode, request.RequestUri);

            //Does the response have any content to log?

            if (exc.Response.ContentLength > 0)

            {

                var content = LogContent(exc.Response);

                Log.ErrorFormat(content);

            }

            return false;

        }

        private readonly List<HttpStatusCode> _allowedStatusCodes=new List<HttpStatusCode>{HttpStatusCode.NotModified};        

        public List<HttpStatusCode> AllowedStatusCodes

        {

            get

            {

                return _allowedStatusCodes;

            }            

        }

        public DateTime LastModified { get; private set; }

        private static string LogContent(WebResponse webResponse)

        {

            if (webResponse == null)

                throw new ArgumentNullException("webResponse");

            Contract.EndContractBlock();

            //The response stream must be copied to avoid affecting other code by disposing of the 

            //original response stream.

            var stream = webResponse.GetResponseStream();            

            using(var memStream=new MemoryStream())

            using (var reader = new StreamReader(memStream))

            {

                stream.CopyTo(memStream);                

                string content = reader.ReadToEnd();

                stream.Seek(0,SeekOrigin.Begin);

                return content;

            }

        }

        public string DownloadStringWithRetry(string address,int retries=0)

        {

            if (address == null)

                throw new ArgumentNullException("address");

            var actualAddress = GetActualAddress(address);

            TraceStart("GET",actualAddress);            

            var actualRetries = (retries == 0) ? Retries : retries;

            var uriString = String.Join("/", BaseAddress.TrimEnd('/'), actualAddress);

            var task = Retry(() =>

            {                                

                var content = DownloadString(uriString);

                if (StatusCode == HttpStatusCode.NoContent)

                    return String.Empty;

                return content;

            }, actualRetries);

            try

            {

                    var result = task.Result;

                return result;

            }

            catch (AggregateException exc)

            {

                //If the task fails, propagate the original exception

                if (exc.InnerException!=null)

                    throw exc.InnerException;

                throw;

            }

        }

        public void Head(string address,int retries=0)

        {

            AllowedStatusCodes.Add(HttpStatusCode.NotFound);

            RetryWithoutContent(address, retries, "HEAD");

        }

        public void PutWithRetry(string address, int retries = 0, string contentType=null)

        {

            RetryWithoutContent(address, retries, "PUT",contentType);

        }

        public void PostWithRetry(string address,string contentType)

        {            

            RetryWithoutContent(address, 0, "POST",contentType);

        }

        public void DeleteWithRetry(string address,int retries=0)

        {

            RetryWithoutContent(address, retries, "DELETE");

        }

        public string GetHeaderValue(string headerName,bool optional=false)

        {

            if (this.ResponseHeaders==null)

                throw new InvalidOperationException("ResponseHeaders are null");

            Contract.EndContractBlock();

            var values=this.ResponseHeaders.GetValues(headerName);

            if (values != null)

                return values[0];

            if (optional)            

                return null;            

            //A required header was not found

            throw new WebException(String.Format("The {0}  header is missing", headerName));

        }

        public void SetNonEmptyHeaderValue(string headerName, string value)

        {

            if (String.IsNullOrWhiteSpace(value))

                return;

            Headers.Add(headerName,value);

        }

        private void RetryWithoutContent(string address, int retries, string method,string contentType=null)

        {

            if (address == null)

                throw new ArgumentNullException("address");

            var actualAddress = GetActualAddress(address);            

            var actualRetries = (retries == 0) ? Retries : retries;

            var task = Retry(() =>

            {

                var uriString = String.Join("/",BaseAddress ,actualAddress);

                var uri = new Uri(uriString);

                var request =  GetWebRequest(uri);

                if (contentType!=null)

                {

                    request.ContentType = contentType;

                    request.ContentLength = 0;

                }

                request.Method = method;

                if (ResponseHeaders!=null)

                    ResponseHeaders.Clear();

                TraceStart(method, uriString);

                if (method == "PUT")

                    request.ContentLength = 0;

                //Have to use try/finally instead of using here, because WebClient needs a valid WebResponse object

                //in order to return response headers

                var response = (HttpWebResponse)GetWebResponse(request);

                try

                {

                    LastModified = response.LastModified;

                    StatusCode = response.StatusCode;

                    StatusDescription = response.StatusDescription;

                }

                finally

                {

                    response.Close();

                }

                return 0;

            }, actualRetries);

            try

            {

                task.Wait();

            }

            catch (AggregateException ex)

            {

                var exc = ex.InnerException;

                if (exc is RetryException)

                {

                    Log.ErrorFormat("[{0}] RETRY FAILED for {1} after {2} retries",method,address,retries);

                }

                else

                {

                    Log.ErrorFormat("[{0}] FAILED for {1} with \n{2}", method, address, exc);

                }

                throw exc;

            }

            catch(Exception ex)

            {

                Log.ErrorFormat("[{0}] FAILED for {1} with \n{2}", method, address, ex);

                throw;

            }

        }

        private static void TraceStart(string method, string actualAddress)

        {

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

        }

        private string GetActualAddress(string address)

        {

            if (Parameters.Count == 0)

                return address;

            var addressBuilder=new StringBuilder(address);            

            bool isFirst = true;

            foreach (var parameter in Parameters)

            {

                if(isFirst)

                    addressBuilder.AppendFormat("?{0}={1}", parameter.Key, parameter.Value);

                else

                    addressBuilder.AppendFormat("&{0}={1}", parameter.Key, parameter.Value);

                isFirst = false;

            }

            return addressBuilder.ToString();

        }

        public string DownloadStringWithRetry(Uri address,int retries=0)

        {

            if (address == null)

                throw new ArgumentNullException("address");

            var actualRetries = (retries == 0) ? Retries : retries;            

            var task = Retry(() =>

            {

                var content = base.DownloadString(address);

                if (StatusCode == HttpStatusCode.NoContent)

                    return String.Empty;

                return content;

            }, actualRetries);

            var result = task.Result;

            return result;

        }

        /// <summary>

        /// Copies headers from another RestClient

        /// </summary>

        /// <param name="source">The RestClient from which the headers are copied</param>

        public void CopyHeaders(RestClient source)

        {

            if (source == null)

                throw new ArgumentNullException("source", "source can't be null");

            Contract.EndContractBlock();

            //The Headers getter initializes the property, it is never null

            Contract.Assume(Headers!=null);

            CopyHeaders(source.Headers,Headers);

        }

        /// <summary>

        /// Copies headers from one header collection to another

        /// </summary>

        /// <param name="source">The source collection from which the headers are copied</param>

        /// <param name="target">The target collection to which the headers are copied</param>

        public static void CopyHeaders(WebHeaderCollection source,WebHeaderCollection target)

        {

            if (source == null)

                throw new ArgumentNullException("source", "source can't be null");

            if (target == null)

                throw new ArgumentNullException("target", "target can't be null");

            Contract.EndContractBlock();

            for (int i = 0; i < source.Count; i++)

            {

                target.Add(source.GetKey(i), source[i]);

            }            

        }

        public void AssertStatusOK(string message)

        {

            if (StatusCode >= HttpStatusCode.BadRequest)

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

        }

        private Task<T> Retry<T>(Func<T> original, int retryCount, TaskCompletionSource<T> tcs = null)

        {

            if (original==null)

                throw new ArgumentNullException("original");

            Contract.EndContractBlock();

            if (tcs == null)

                tcs = new TaskCompletionSource<T>();

            Task.Factory.StartNew(original).ContinueWith(_original =>

                {

                    if (!_original.IsFaulted)

                        tcs.SetFromTask(_original);

                    else 

                    {

                        var e = _original.Exception.InnerException;

                        var we = (e as WebException);

                        if (we==null)

                            tcs.SetException(e);

                        else

                        {

                            var statusCode = GetStatusCode(we);

                            //Return null for 404

                            if (statusCode == HttpStatusCode.NotFound)

                                tcs.SetResult(default(T));

                            //Retry for timeouts and service unavailable

                            else if (we.Status == WebExceptionStatus.Timeout ||

                                (we.Status == WebExceptionStatus.ProtocolError && statusCode == HttpStatusCode.ServiceUnavailable))

                            {

                                TimedOut = true;

                                if (retryCount == 0)

                                {                                    

                                    Log.ErrorFormat("[ERROR] Timed out too many times. \n{0}\n",e);

                                    tcs.SetException(new RetryException("Timed out too many times.", e));                                    

                                }

                                else

                                {

                                    Log.ErrorFormat(

                                        "[RETRY] Timed out after {0} ms. Will retry {1} more times\n{2}", Timeout,

                                        retryCount, e);

                                    Retry(original, retryCount - 1, tcs);

                                }

                            }

                            else

                                tcs.SetException(e);

                        }

                    };

                });

            return tcs.Task;

        }

        private HttpStatusCode GetStatusCode(WebException we)

        {

            if (we==null)

                throw new ArgumentNullException("we");

            var statusCode = HttpStatusCode.RequestTimeout;

            if (we.Response != null)

            {

                statusCode = ((HttpWebResponse) we.Response).StatusCode;

                this.StatusCode = statusCode;

            }

            return statusCode;

        }

        public UriBuilder GetAddressBuilder(string container, string objectName)

        {

            var builder = new UriBuilder(String.Join("/", BaseAddress, container, objectName));

            return builder;

        }

        public Dictionary<string, string> GetMeta(string metaPrefix)

        {

            if (String.IsNullOrWhiteSpace(metaPrefix))

                throw new ArgumentNullException("metaPrefix");

            Contract.EndContractBlock();

            var keys = ResponseHeaders.AllKeys.AsQueryable();

            var dict = (from key in keys

                        where key.StartsWith(metaPrefix)

                        let name = key.Substring(metaPrefix.Length)

                        select new { Name = name, Value = ResponseHeaders[key] })

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

            return dict;

        }

        internal Task DownloadFileTaskAsync(Uri uri, string fileName, CancellationToken cancellationToken, IProgress<DownloadProgressChangedEventArgs> progress)

        {

            cancellationToken.Register(CancelAsync);

            DownloadProgressChangedEventHandler onDownloadProgressChanged = (o, e) => progress.Report(e);

            this.DownloadProgressChanged += onDownloadProgressChanged;

            return this.DownloadFileTaskAsync(uri, fileName).ContinueWith(t=>

                {

                    this.DownloadProgressChanged -= onDownloadProgressChanged;

                });

        }

        internal Task<byte[]> DownloadDataTaskAsync(Uri uri, CancellationToken cancellationToken, IProgress<DownloadProgressChangedEventArgs> progress)

        {

            cancellationToken.Register(CancelAsync);

            DownloadProgressChangedEventHandler onDownloadProgressChanged = (o, e) => progress.Report(e);

            this.DownloadProgressChanged += onDownloadProgressChanged;

            return this.DownloadDataTaskAsync(uri).ContinueWith(t =>

            {

                this.DownloadProgressChanged -= onDownloadProgressChanged;

                return t.Result;

            });

        }

    }

    public class RetryException:Exception

    {

        public RetryException()

            :base()

        {

        }

        public RetryException(string message)

            :base(message)

        {

        }

        public RetryException(string message,Exception innerException)

            :base(message,innerException)

        {

        }

        public RetryException(SerializationInfo info,StreamingContext context)

            :base(info,context)

        {

        }

    }

}