code.grnet.gr Git - pithos-ms-client/blob - trunk/Pithos.Core/TaskExtensions.cs

   1 using System;
   2 using System.Collections.Concurrent;
   3 using System.Collections.Generic;
   4 using System.Diagnostics.Contracts;
   5 using System.Linq;
   6 using System.Text;
   7 using System.Threading;
   8 using System.Threading.Tasks;
   9
  10 namespace Pithos.Core
  11 {
  12     public static class TaskExtensions
  13     {
  14         public static Task<T2> Then<T1, T2>(this Task<T1> first, Func<T1, Task<T2>> next)
  15         {
  16             if (first == null)
  17                 throw new ArgumentNullException("first");
  18             if (next == null)
  19                 throw new ArgumentNullException("next");
  20             Contract.EndContractBlock();
  21             return Then(first, next, CancellationToken.None);
  22         }
  23
  24         public static Task Then<T1>(this Task<T1> first, Func<T1, Task> next)
  25         {
  26             if (first == null)
  27                 throw new ArgumentNullException("first");
  28             if (next == null)
  29                 throw new ArgumentNullException("next");
  30             Contract.EndContractBlock();
  31             return Then(first, next, CancellationToken.None);
  32         }
  33
  34         public static Task<T2> Then<T1, T2>(this Task<T1> first, Func<T1, Task<T2>> next, CancellationToken cancellationToken)
  35         {
  36             if (first == null)
  37                 throw new ArgumentNullException("first");
  38             if (next == null)
  39                 throw new ArgumentNullException("next");
  40             Contract.EndContractBlock();
  41             Contract.Assume(TaskScheduler.Current!=null);
  42
  43             var tcs = new TaskCompletionSource<T2>();
  44             first.ContinueWith(delegate
  45             {
  46                 if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions);
  47                 else if (first.IsCanceled) tcs.TrySetCanceled();
  48                 else
  49                 {
  50                     try
  51                     {
  52                         var t = next(first.Result);
  53                         if (t == null) tcs.TrySetCanceled();
  54                         else t.ContinueWith(delegate
  55                         {
  56                             if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions);
  57                             else if (t.IsCanceled) tcs.TrySetCanceled();
  58                             else tcs.TrySetResult(t.Result);
  59                         }, TaskContinuationOptions.ExecuteSynchronously);
  60                     }
  61                     catch (Exception exc) { tcs.TrySetException(exc); }
  62                 }
  63             }, cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Current);
  64             return tcs.Task;
  65         }
  66
  67         public static Task Then<T1>(this Task<T1> first, Func<T1, Task> next, CancellationToken cancellationToken)
  68         {
  69             if (first == null)
  70                 throw new ArgumentNullException("first");
  71             if (next == null)
  72                 throw new ArgumentNullException("next");
  73             Contract.EndContractBlock();
  74             Contract.Assume(TaskScheduler.Current != null);
  75
  76             var tcs = new TaskCompletionSource<object>();
  77             first.ContinueWith(delegate
  78             {
  79                 if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions);
  80                 else if (first.IsCanceled) tcs.TrySetCanceled();
  81                 else
  82                 {
  83                     try
  84                     {
  85                         var t = next(first.Result);
  86                         if (t == null) tcs.TrySetCanceled();
  87                         else t.ContinueWith(delegate
  88                         {
  89                             if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions);
  90                             else if (t.IsCanceled) tcs.TrySetCanceled();
  91                             else tcs.TrySetResult(null);
  92                         }, TaskContinuationOptions.ExecuteSynchronously);
  93                     }
  94                     catch (Exception exc) { tcs.TrySetException(exc); }
  95                 }
  96             }, cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Current);
  97             return tcs.Task;
  98         }
  99
 100
 101
 102         public static void ReportExceptions(this Task task,Action<AggregateException> OnError,Action OnSuccess )
 103         {
 104             if (!task.IsCompleted) throw new InvalidOperationException("The task has not completed.");
 105             if (task.IsFaulted)
 106                 task.Exception.Handle(exc=>
 107                                           {
 108                                               OnError(task.Exception);
 109                                               return true;
 110                                           });
 111             else
 112             {
 113                 OnSuccess();
 114             }
 115         }
 116
 117         public static bool TryRemove<T>(this ObservableConcurrentCollection<T> collection,T item) where T:class
 118         {
 119             IProducerConsumerCollection<T> items= collection;
 120             for (var i = 0; i < items.Count; i++)
 121             {
 122                 T tempItem;
 123                 if (!items.TryTake(out tempItem))
 124                     return false;
 125                 if (tempItem == item)
 126                     return true;
 127                 items.TryAdd(item);
 128             }
 129             return false;
 130         }
 131
 132     }
 133 }