##// END OF EJS Templates
implab » ImplabNet
RSS
DRAFT: refactoring, adding cancelation token
DRAFT: refactoring, adding cancelation token
cin - - Load All Authors

File last commit:

r138:f75cfa58e3d4 v2
r143:16f926ee499d v2
Show More
PromiseExtensions.cs
192 lines | 6.4 KiB | text/x-csharp | CSharpLexer
/ Implab / PromiseExtensions.cs
History | Source | Raw |Copy content |Copy permalink
cin
promises refactoring
 r72 using System.Threading;
cin
major refactoring, added tasks support
 r75 using System;
cin
minor fixes
 r109 using Implab.Diagnostics;
cin
Promises rewritten, added improved version of AsyncQueue
 r119 using System.Collections.Generic;
cin
minor fixes
 r109
cin
major refactoring, added tasks support
 r75 #if NET_4_5
using System.Threading.Tasks;
#endif
cin
promises refactoring
 r72
namespace Implab {
public static class PromiseExtensions {
public static IPromise<T> DispatchToCurrentContext<T>(this IPromise<T> that) {
cin
major refactoring, added tasks support
 r75 Safe.ArgumentNotNull(that, "that");
cin
promises refactoring
 r72 var context = SynchronizationContext.Current;
if (context == null)
return that;
cin
Promises rewritten, added improved version of AsyncQueue
 r119 var p = new SyncContextPromise<T>(context);
p.On(that.Cancel, PromiseEventType.Cancelled);
cin
promises refactoring
 r72
cin
renamed Promise.Last -> Promise.On...
 r104 that.On(
cin
Refactoring of the IPromise<T> interface...
 r76 p.Resolve,
p.Reject,
p.Cancel
cin
promises refactoring
 r72 );
return p;
}
public static IPromise<T> DispatchToContext<T>(this IPromise<T> that, SynchronizationContext context) {
cin
major refactoring, added tasks support
 r75 Safe.ArgumentNotNull(that, "that");
cin
promises refactoring
 r72 Safe.ArgumentNotNull(context, "context");
cin
Promises rewritten, added improved version of AsyncQueue
 r119 var p = new SyncContextPromise<T>(context);
p.On(that.Cancel, PromiseEventType.Cancelled);
cin
promises refactoring
 r72
cin
renamed Promise.Last -> Promise.On...
 r104 that.On(
cin
Refactoring of the IPromise<T> interface...
 r76 p.Resolve,
p.Reject,
p.Cancel
cin
promises refactoring
 r72 );
return p;
}
cin
major refactoring, added tasks support
 r75
cin
code cleanup...
 r101 /// <summary>
/// Ensures the dispatched.
/// </summary>
/// <returns>The dispatched.</returns>
/// <param name="that">That.</param>
/// <param name="head">Head.</param>
/// <param name="cleanup">Cleanup.</param>
/// <typeparam name="TPromise">The 1st type parameter.</typeparam>
/// <typeparam name="T">The 2nd type parameter.</typeparam>
public static TPromise EnsureDispatched<TPromise,T>(this TPromise that, IPromise<T> head, Action<T> cleanup) where TPromise : IPromise{
Safe.ArgumentNotNull(that, "that");
Safe.ArgumentNotNull(head, "head");
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 that.On(() => head.On(cleanup), PromiseEventType.Cancelled);
cin
code cleanup...
 r101
return that;
}
cin
major refactoring, added tasks support
 r75 public static AsyncCallback AsyncCallback<T>(this Promise<T> that, Func<IAsyncResult,T> callback) {
Safe.ArgumentNotNull(that, "that");
Safe.ArgumentNotNull(callback, "callback");
cin
minor fixes
 r109 var op = TraceContext.Instance.CurrentOperation;
cin
major refactoring, added tasks support
 r75 return ar => {
cin
minor fixes
 r109 TraceContext.Instance.EnterLogicalOperation(op,false);
cin
major refactoring, added tasks support
 r75 try {
that.Resolve(callback(ar));
} catch (Exception err) {
that.Reject(err);
cin
minor fixes
 r109 } finally {
TraceContext.Instance.Leave();
cin
major refactoring, added tasks support
 r75 }
};
}
cin
added promise timeout helper
 r110
static void CancelCallback(object cookie) {
((ICancellable)cookie).Cancel();
}
/// <summary>
/// Cancells promise after the specified timeout is elapsed.
/// </summary>
/// <param name="that">The promise to cancel on timeout.</param>
/// <param name="milliseconds">The timeout in milliseconds.</param>
/// <typeparam name="TPromise">The 1st type parameter.</typeparam>
public static TPromise Timeout<TPromise>(this TPromise that, int milliseconds) where TPromise : IPromise {
Safe.ArgumentNotNull(that, "that");
var timer = new Timer(CancelCallback, that, milliseconds, -1);
that.On(timer.Dispose, PromiseEventType.All);
return that;
}
cin
Promises rewritten, added improved version of AsyncQueue
 r119
cin
major update, added Drain mathod to AsyncQueue class
 r124 public static IPromise Bundle(this ICollection<IPromise> that) {
cin
Promises rewritten, added improved version of AsyncQueue
 r119 Safe.ArgumentNotNull(that, "that");
int count = that.Count;
cin
major update, added Drain mathod to AsyncQueue class
 r124 int errors = 0;
cin
Promises rewritten, added improved version of AsyncQueue
 r119 var medium = new Promise();
cin
shared locks + tests
 r136 if (count == 0) {
medium.Resolve();
return medium;
}
cin
major update, added Drain mathod to AsyncQueue class
 r124 medium.On(() => {
foreach(var p2 in that)
p2.Cancel();
}, PromiseEventType.ErrorOrCancel);
cin
Promises rewritten, added improved version of AsyncQueue
 r119 foreach (var p in that)
p.On(
() => {
if (Interlocked.Decrement(ref count) == 0)
medium.Resolve();
},
error => {
cin
major update, added Drain mathod to AsyncQueue class
 r124 if (Interlocked.Increment(ref errors) == 1)
medium.Reject(
new Exception("The dependency promise is failed", error)
);
cin
Promises rewritten, added improved version of AsyncQueue
 r119 },
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 reason => {
cin
major update, added Drain mathod to AsyncQueue class
 r124 if (Interlocked.Increment(ref errors) == 1)
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 medium.Cancel(
cin
major update, added Drain mathod to AsyncQueue class
 r124 new Exception("The dependency promise is cancelled")
);
cin
Promises rewritten, added improved version of AsyncQueue
 r119 }
);
return medium;
}
cin
major update, added Drain mathod to AsyncQueue class
 r124
public static IPromise<T[]> Bundle<T>(this ICollection<IPromise<T>> that) {
Safe.ArgumentNotNull(that, "that");
int count = that.Count;
int errors = 0;
var medium = new Promise<T[]>();
var results = new T[that.Count];
medium.On(() => {
foreach(var p2 in that)
p2.Cancel();
}, PromiseEventType.ErrorOrCancel);
int i = 0;
foreach (var p in that) {
var idx = i;
p.On(
x => {
results[idx] = x;
if (Interlocked.Decrement(ref count) == 0)
medium.Resolve(results);
},
error => {
if (Interlocked.Increment(ref errors) == 1)
medium.Reject(
new Exception("The dependency promise is failed", error)
);
},
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 reason => {
cin
major update, added Drain mathod to AsyncQueue class
 r124 if (Interlocked.Increment(ref errors) == 1)
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 medium.Cancel(
new Exception("The dependency promise is cancelled", reason)
cin
major update, added Drain mathod to AsyncQueue class
 r124 );
}
);
i++;
}
return medium;
}
cin
major refactoring, added tasks support
 r75
#if NET_4_5
public static Task<T> GetTask<T>(this IPromise<T> that) {
Safe.ArgumentNotNull(that, "that");
var tcs = new TaskCompletionSource<T>();
cin
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
 r138 that.On(tcs.SetResult, tcs.SetException, r => tcs.SetCanceled());
cin
major refactoring, added tasks support
 r75
return tcs.Task;
}
#endif
cin
promises refactoring
 r72 }
}