Synnefo » Pithos MS Client

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Diagnostics.Contracts;

using System.Linq;

using System.Net.Http;

using System.Net.Http.Headers;

using System.Reflection;

using System.Text;

using System.Net;

using System.IO;

using System.Threading;

using System.Threading.Tasks;

using Pithos.Interfaces;

using log4net;

using System.ServiceModel.Channels;

namespace Pithos.Network

{

    public static class WebExtensions

    {

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

        public static string ReadToEnd(this HttpWebResponse response)

        {

            using (var stream = response.GetResponseStream())

            {

                if (stream == null)

                    return null;

                using (var reader = new StreamReader(stream))

                {

                    var body = reader.ReadToEnd();

                    return body;

                }

            }

        }

        public static void LogError(this ILog log, HttpWebResponse response)

        {

            if (log.IsDebugEnabled)

            {

                if (response != null)

                {

                    var body = response.ReadToEnd();

                    log.ErrorFormat("Headers:\n{0}\nBody:{1}", response.Headers, body);

                }

            }

        }

        public static TextReader GetLoggedReader(this Stream stream, ILog log)

        {

            var reader = new StreamReader(stream);

            if (!log.IsDebugEnabled)

                return reader;

            using (reader)

            {

                var body = reader.ReadToEnd();

                log.DebugFormat("JSON response: {0}", body);

                return new StringReader(body);

            }

        }

        public static IEnumerable<T> Range<T>(this IList<T> list, int start, int end)

        {

            Contract.Requires(start >= 0);

            Contract.Requires(end < list.Count);

            Contract.Requires(start <= end);

            if (list == null)

                yield break;

            for (var i = 0; i <= end; i++)

            {

                yield return list[i];

            }

        }

        public static Task<byte[]> UploadDataTaskAsync(this WebClient webClient, Uri address, string method, byte[] data, CancellationToken cancellationToken, IProgress<UploadProgressChangedEventArgs> progress)

        {

            var tcs = new TaskCompletionSource<byte[]>(address);

            if (cancellationToken.IsCancellationRequested)

            {

                tcs.TrySetCanceled();

            }

            else

            {

                CancellationTokenRegistration ctr = cancellationToken.Register(()=>

                {

                    webClient.CancelAsync();

                });

                UploadDataCompletedEventHandler completedHandler = null;

                UploadProgressChangedEventHandler progressHandler = null;

                if (progress != null)

                    progressHandler = (s, e) => PithosEAPCommon.HandleProgress(tcs, e, () => e, progress);

                completedHandler =(sender, e) =>PithosEAPCommon.HandleCompletion(tcs, true, e,() => e.Result,() =>

                { 

                    ctr.Dispose();

                    webClient.UploadDataCompleted -= completedHandler;

                    webClient.UploadProgressChanged -= progressHandler;

                });

                webClient.UploadDataCompleted += completedHandler;

                webClient.UploadProgressChanged += progressHandler;

                try

                {

                    webClient.UploadDataAsync(address, method, data, tcs);

                    if (cancellationToken.IsCancellationRequested)

                        webClient.CancelAsync();

                }

                catch

                {

                    webClient.UploadDataCompleted -= completedHandler;

                    webClient.UploadProgressChanged -= progressHandler;

                    throw;

                }

            }

            return tcs.Task;

        }

        public static async Task<T> WithRetries<T>(this Func<Task<T>> func, int retries)

        {

            while (retries > 0)

            {

                try

                {

                    var result = await func().ConfigureAwait(false);

                    return result;

                }

                catch (Exception exc)

                {

                    if (--retries == 0)

                        throw new RetryException("Failed too many times", exc);

                }

            }

            throw new RetryException();

        }

        public static async Task<HttpResponseMessage> WithRetriesForWeb(this Func<Task<HttpResponseMessage>> func, int retries)

        {

            var waitTime = TimeSpan.FromSeconds(10);

            var acceptedCodes = new[] { HttpStatusCode.Moved, HttpStatusCode.MovedPermanently, HttpStatusCode.Found, HttpStatusCode.Redirect,HttpStatusCode.SeeOther,

                HttpStatusCode.RedirectMethod,HttpStatusCode.NotModified,HttpStatusCode.TemporaryRedirect,HttpStatusCode.RedirectKeepVerb};

            while (retries > 0)

            {

                var result = await func().ConfigureAwait(false);

                if (result.IsSuccessStatusCode || acceptedCodes.Contains(result.StatusCode))

                    return result;

                if (--retries == 0)

                    throw new RetryException("Failed too many times");

                //Wait for service unavailable

                if (result.StatusCode == HttpStatusCode.ServiceUnavailable)

                {

                    Log.InfoFormat("[UNAVAILABLE] Waiting before retry: {0}",result.ReasonPhrase);

                    await TaskEx.Delay(waitTime).ConfigureAwait(false);

                    //increase the timeout for repeated timeouts

                    if (waitTime<TimeSpan.FromSeconds(10))

                        waitTime = waitTime.Add(TimeSpan.FromSeconds(10));

                }                

                //Throw in all other cases

                else 

                    result.EnsureSuccessStatusCode();

            }

            throw new RetryException();

        }

        public static async Task<string> GetStringAsyncWithRetries(this HttpClient client, Uri requestUri, int retries,DateTime? since=null)

        {                        

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

            if (since.HasValue)

            {

                request.Headers.IfModifiedSince = since.Value;

            }

            //Func<Task<HttpResponseMessage>> call = () => _baseHttpClient.SendAsync(request);

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

            {

                if (response.StatusCode == HttpStatusCode.NoContent)

                    return String.Empty;

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

                return content;

            }

        }

        public static Task<HttpResponseMessage> HeadAsyncWithRetries(this HttpClient client, Uri requestUri, int retries,bool acceptNotFound=false)

        {

            return client.HeadAsyncWithRetries(requestUri, retries, acceptNotFound,HttpCompletionOption.ResponseContentRead, CancellationToken.None);

        }

        public static Task<HttpResponseMessage> HeadAsyncWithRetries(this HttpClient client, Uri requestUri, int retries, bool acceptNotFound,HttpCompletionOption completionOption, CancellationToken cancellationToken)

        {

            return client.SendAsyncWithRetries(new HttpRequestMessage(HttpMethod.Head, requestUri), retries, acceptNotFound,completionOption, cancellationToken);

        }

        public static Task<HttpResponseMessage> GetAsyncWithRetries(this HttpClient client, Uri requestUri, int retries)

        {

            return client.GetAsyncWithRetries(requestUri, retries, HttpCompletionOption.ResponseContentRead, CancellationToken.None);

        }

        public static Task<HttpResponseMessage> GetAsyncWithRetries(this HttpClient client, Uri requestUri, int retries, HttpCompletionOption completionOption, CancellationToken cancellationToken)

        {

            return client.SendAsyncWithRetries(new HttpRequestMessage(HttpMethod.Get, requestUri), retries, false,completionOption, cancellationToken);

        }

        public static Task<HttpResponseMessage> SendAsyncWithRetries(this HttpClient client,HttpRequestMessage message, int retries)

        {

            return client.SendAsyncWithRetries(message, retries,false, HttpCompletionOption.ResponseContentRead, CancellationToken.None);

        }

        public static async Task<HttpResponseMessage> SendAsyncWithRetries(this HttpClient client,HttpRequestMessage message, int retries,bool acceptNotFound,HttpCompletionOption completionOption, CancellationToken cancellationToken)

        {

            var waitTime = TimeSpan.FromSeconds(10);

            var acceptedCodes = new HashSet<HttpStatusCode> { HttpStatusCode.Moved, HttpStatusCode.MovedPermanently, 

                HttpStatusCode.Found, HttpStatusCode.Redirect,HttpStatusCode.SeeOther,HttpStatusCode.RedirectMethod,

                HttpStatusCode.NotModified,HttpStatusCode.TemporaryRedirect,HttpStatusCode.RedirectKeepVerb,

                HttpStatusCode.Conflict};

            if (acceptNotFound)

                acceptedCodes.Add(HttpStatusCode.NotFound);

            while (retries > 0)

            {

                var timedOut = false;

                if (Log.IsDebugEnabled)

                    Log.DebugFormat("[REQUEST] {0}", message);

                HttpResponseMessage result=null;

                var msg = message.Clone();

                Exception innerException=null;

                try

                {

                    result = await client.SendAsync(msg, completionOption, cancellationToken).ConfigureAwait(false);

                    if (result.IsSuccessStatusCode || acceptedCodes.Contains(result.StatusCode))

                    {

                        if (Log.IsDebugEnabled)

                            Log.DebugFormat("[RESPONSE] [{0}]:[{1}] OK: [{2}]", msg.Method, msg.RequestUri,

                                            result.StatusCode);

                        return result;

                    }

                }

                catch (HttpRequestException exc)

                {

                    //A timeout or other error caused a failure without receiving a response from the server

                    if (Log.IsDebugEnabled)

                        Log.DebugFormat("[RESPONSE] [{0}]:[{1}] HTTP FAIL :\n{2}", msg.Method, msg.RequestUri,exc);

                    innerException = exc;

                }

                catch (WebException exc)

                {

                    if (exc.Status != WebExceptionStatus.Timeout)

                        throw;

                    timedOut = true;

                    if (Log.IsDebugEnabled)

                        Log.DebugFormat("[RESPONSE] [{0}]:[{1}] FAIL WITH TIMEOUT", msg.Method, msg.RequestUri);

                    innerException = exc;

                }

                catch(TaskCanceledException exc)

                {

                    //If the task was cancelled due to a timeout, retry it

                    if (!exc.CancellationToken.IsCancellationRequested)

                    {

                        timedOut = true;

                        if (Log.IsDebugEnabled)

                            Log.DebugFormat("[RESPONSE] [{0}]:[{1}] FAIL WITH TIMEOUT", msg.Method, msg.RequestUri);

                    }

                    else

                    {

                        throw;

                    }

                    innerException = exc;

                }

                catch (Exception exc)

                {

                    Log.FatalFormat("Unexpected error while sending:\n{0}\n{1}", msg, exc);

                    throw;

                }

                //Report the failure

                var resultStatus = result.NullSafe(s => s.StatusCode);

                if (Log.IsDebugEnabled)

                {                    

                    if (timedOut)

                        Log.DebugFormat("[TIMEOUT] [{0}]:[{1}]", msg.Method, msg.RequestUri);

                    else

                        Log.DebugFormat("[RESPONSE] [{0}]:[{1}] FAIL: [{2}]\n{3}", msg.Method, msg.RequestUri,

                                        resultStatus, result);

                }

                //Handle the failure

                if (--retries == 0)

                    throw new RetryException("Failed too many times");

                //Wait for service unavailable

                if (resultStatus == HttpStatusCode.ServiceUnavailable ||

                    resultStatus == HttpStatusCode.BadGateway ||

                    resultStatus == HttpStatusCode.InternalServerError )

                {

                    Log.WarnFormat("[UNAVAILABLE] Waiting before retrying [{0}]:[{1}] due to [{2}]", msg.Method,

                                   msg.RequestUri, result.ReasonPhrase);

                    await TaskEx.Delay(waitTime).ConfigureAwait(false);

                    //increase the timeout for repeated timeouts

                    if (waitTime < TimeSpan.FromSeconds(10))

                        waitTime = waitTime.Add(TimeSpan.FromSeconds(10));

                }                           

                //Throw in all other cases, unless there was a client-side timeout

                else if (!timedOut)

                    //Throw if there was no result (never received a reply)

                    if (result == null)

                        throw innerException ?? new RetryException();                        

                    else

                        //Otherwise force the exception that corresponds to the response

                        throw new HttpRequestWithStatusException(result.StatusCode,result.ReasonPhrase);

            }

            throw new RetryException();

        }

        public static HttpRequestMessage Clone(this HttpRequestMessage message)

        {

            var newMessage = new HttpRequestMessage(message.Method, message.RequestUri);

            foreach (var header in message.Headers)

            {

                newMessage.Headers.Add(header.Key, header.Value);

            }

            newMessage.Content = message.Content.Clone();

            foreach (var property in message.Properties)

            {

                newMessage.Properties.Add(property.Key, property.Value);

            }

            return newMessage;

        }

        public static HttpContent Clone(this HttpContent content)

        {

            if (content == null)

                return null;

            if (!(content is FileBlockContent || 

                content is ByteArrayContentWithProgress || 

                content is StringContent ||

                content is ByteArrayContent))

                throw new ArgumentException("content");

            HttpContent newContent=null;

            if (content is FileBlockContent)

            {

                newContent = (content as FileBlockContent).Clone();

            }

            else if (content is ByteArrayContentWithProgress)

            {

                newContent=(content as ByteArrayContentWithProgress).Clone();

            }

            else if (content is StringContent)

            {

                var fieldInfo = typeof(ByteArrayContent).GetField("content",

                                                                    BindingFlags.NonPublic | BindingFlags.Instance);

                var bytes = (byte[])fieldInfo.GetValue(content);

                var enc=Encoding.GetEncoding(content.Headers.ContentType.CharSet);

                var stringContent=enc.GetString(bytes);

                newContent = new StringContent(stringContent);

            }

            else if (content is ByteArrayContent)

            {

                var fieldInfo = typeof(ByteArrayContent).GetField("content",

                                                                    BindingFlags.NonPublic | BindingFlags.Instance);

                var bytes = (byte[])fieldInfo.GetValue(content);

                newContent = new ByteArrayContent(bytes);

            }

            foreach (var header in content.Headers)

            {

                if (!(header.Key == "Content-Type" && newContent.Headers.Contains(header.Key)))

                    newContent.Headers.Add(header.Key, header.Value);

            }

            return newContent;

        }

        public static string GetFirstValue(this HttpResponseHeaders headers, string name)

        {

            if (headers==null)

                throw new ArgumentNullException("headers");

            if (String.IsNullOrWhiteSpace(name))

                throw new ArgumentNullException("name");

            Contract.EndContractBlock();

            IEnumerable<string> values;

            if (headers.TryGetValues(name, out values))

            {

                return values.FirstOrDefault();

            }

            return null;

        }

        public static  Dictionary<string, string> GetMeta(this HttpResponseHeaders headers,string metaPrefix)

        {

            if (headers == null)

                throw new ArgumentNullException("headers");

            if (String.IsNullOrWhiteSpace(metaPrefix))

                throw new ArgumentNullException("metaPrefix");

            Contract.EndContractBlock();

            var dict = (from header in headers

                        where header.Key.StartsWith(metaPrefix)

                        let name = header.Key.Substring(metaPrefix.Length)

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

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

            return dict;

        }

    }

    internal static class PithosEAPCommon

    {

        internal static void HandleProgress<T, E>(TaskCompletionSource<T> tcs, ProgressChangedEventArgs eventArgs, Func<E> getProgress, IProgress<E> callback)

        {

            if (eventArgs.UserState != tcs)

                return;

            callback.Report(getProgress());

        }

        internal static void HandleCompletion<T>(TaskCompletionSource<T> tcs, bool requireMatch, AsyncCompletedEventArgs e, Func<T> getResult, Action unregisterHandler)

        {

            if (requireMatch)

            {

                if (e.UserState != tcs)

                    return;

            }

            try

            {

                unregisterHandler();

            }

            finally

            {

                if (e.Cancelled)

                    tcs.TrySetCanceled();

                else if (e.Error != null)

                    tcs.TrySetException(e.Error);

                else

                    tcs.TrySetResult(getResult());

            }

        }

    }

}