Mercurial > pub > ImplabNet

annotate Implab/PromiseExtensions.cs @ 129:471f596b2603 v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Added SharedLock to synchronization routines
author cin
date Thu, 29 Jan 2015 18:31:06 +0300
parents a336cb13c6a9
children e9e7940c7d98
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
1 using System.Threading;
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
2 using System;
109
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
3 using Implab.Diagnostics;
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
4 using System.Collections.Generic;
109
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
5
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
6
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
7 #if NET_4_5
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
8 using System.Threading.Tasks;
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
9 #endif
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
10
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
11 namespace Implab {
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
12 public static class PromiseExtensions {
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
13 public static IPromise<T> DispatchToCurrentContext<T>(this IPromise<T> that) {
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
14 Safe.ArgumentNotNull(that, "that");
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
15 var context = SynchronizationContext.Current;
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
16 if (context == null)
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
17 return that;
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
18
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
19 var p = new SyncContextPromise<T>(context);
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
20 p.On(that.Cancel, PromiseEventType.Cancelled);
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
21
104
5f10d54b45df renamed Promise.Last -> Promise.On
cin
parents: 101
diff changeset
22 that.On(
76
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
23 p.Resolve,
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
24 p.Reject,
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
25 p.Cancel
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
26 );
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
27 return p;
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
28 }
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
29
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
30 public static IPromise<T> DispatchToContext<T>(this IPromise<T> that, SynchronizationContext context) {
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
31 Safe.ArgumentNotNull(that, "that");
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
32 Safe.ArgumentNotNull(context, "context");
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
33
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
34 var p = new SyncContextPromise<T>(context);
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
35 p.On(that.Cancel, PromiseEventType.Cancelled);
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
36
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
37
104
5f10d54b45df renamed Promise.Last -> Promise.On
cin
parents: 101
diff changeset
38 that.On(
76
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
39 p.Resolve,
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
40 p.Reject,
c761fc982e1d Refactoring of the IPromise<T> interface
cin
parents: 75
diff changeset
41 p.Cancel
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
42 );
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
43 return p;
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
44 }
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
45
101
279e226dffdd code cleanup
cin
parents: 76
diff changeset
46 /// <summary>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
47 /// Ensures the dispatched.
279e226dffdd code cleanup
cin
parents: 76
diff changeset
48 /// </summary>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
49 /// <returns>The dispatched.</returns>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
50 /// <param name="that">That.</param>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
51 /// <param name="head">Head.</param>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
52 /// <param name="cleanup">Cleanup.</param>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
53 /// <typeparam name="TPromise">The 1st type parameter.</typeparam>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
54 /// <typeparam name="T">The 2nd type parameter.</typeparam>
279e226dffdd code cleanup
cin
parents: 76
diff changeset
55 public static TPromise EnsureDispatched<TPromise,T>(this TPromise that, IPromise<T> head, Action<T> cleanup) where TPromise : IPromise{
279e226dffdd code cleanup
cin
parents: 76
diff changeset
56 Safe.ArgumentNotNull(that, "that");
279e226dffdd code cleanup
cin
parents: 76
diff changeset
57 Safe.ArgumentNotNull(head, "head");
279e226dffdd code cleanup
cin
parents: 76
diff changeset
58
104
5f10d54b45df renamed Promise.Last -> Promise.On
cin
parents: 101
diff changeset
59 that.On(null,null,() => head.On(cleanup));
101
279e226dffdd code cleanup
cin
parents: 76
diff changeset
60
279e226dffdd code cleanup
cin
parents: 76
diff changeset
61 return that;
279e226dffdd code cleanup
cin
parents: 76
diff changeset
62 }
279e226dffdd code cleanup
cin
parents: 76
diff changeset
63
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
64 public static AsyncCallback AsyncCallback<T>(this Promise<T> that, Func<IAsyncResult,T> callback) {
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
65 Safe.ArgumentNotNull(that, "that");
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
66 Safe.ArgumentNotNull(callback, "callback");
109
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
67 var op = TraceContext.Instance.CurrentOperation;
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
68 return ar => {
109
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
69 TraceContext.Instance.EnterLogicalOperation(op,false);
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
70 try {
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
71 that.Resolve(callback(ar));
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
72 } catch (Exception err) {
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
73 that.Reject(err);
109
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
74 } finally {
1b7ebcc52e5a minor fixes
cin
parents: 104
diff changeset
75 TraceContext.Instance.Leave();
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
76 }
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
77 };
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
78 }
110
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
79
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
80 static void CancelCallback(object cookie) {
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
81 ((ICancellable)cookie).Cancel();
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
82 }
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
83
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
84 /// <summary>
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
85 /// Cancells promise after the specified timeout is elapsed.
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
86 /// </summary>
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
87 /// <param name="that">The promise to cancel on timeout.</param>
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
88 /// <param name="milliseconds">The timeout in milliseconds.</param>
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
89 /// <typeparam name="TPromise">The 1st type parameter.</typeparam>
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
90 public static TPromise Timeout<TPromise>(this TPromise that, int milliseconds) where TPromise : IPromise {
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
91 Safe.ArgumentNotNull(that, "that");
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
92 var timer = new Timer(CancelCallback, that, milliseconds, -1);
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
93 that.On(timer.Dispose, PromiseEventType.All);
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
94 return that;
1a8426e6e895 added promise timeout helper
cin
parents: 109
diff changeset
95 }
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
96
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
97 public static IPromise Bundle(this ICollection<IPromise> that) {
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
98 Safe.ArgumentNotNull(that, "that");
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
99
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
100 int count = that.Count;
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
101 int errors = 0;
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
102 var medium = new Promise();
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
103
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
104 medium.On(() => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
105 foreach(var p2 in that)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
106 p2.Cancel();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
107 }, PromiseEventType.ErrorOrCancel);
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
108
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
109 foreach (var p in that)
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
110 p.On(
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
111 () => {
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
112 if (Interlocked.Decrement(ref count) == 0)
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
113 medium.Resolve();
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
114 },
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
115 error => {
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
116 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
117 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
118 new Exception("The dependency promise is failed", error)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
119 );
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
120 },
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
121 () => {
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
122 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
123 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
124 new Exception("The dependency promise is cancelled")
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
125 );
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
126 }
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
127 );
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
128
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
129 return medium;
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
130 }
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
131
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
132 public static IPromise<T[]> Bundle<T>(this ICollection<IPromise<T>> that) {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
133 Safe.ArgumentNotNull(that, "that");
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
134
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
135 int count = that.Count;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
136 int errors = 0;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
137 var medium = new Promise<T[]>();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
138 var results = new T[that.Count];
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
139
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
140 medium.On(() => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
141 foreach(var p2 in that)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
142 p2.Cancel();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
143 }, PromiseEventType.ErrorOrCancel);
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
144
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
145 int i = 0;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
146 foreach (var p in that) {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
147 var idx = i;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
148 p.On(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
149 x => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
150 results[idx] = x;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
151 if (Interlocked.Decrement(ref count) == 0)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
152 medium.Resolve(results);
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
153 },
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
154 error => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
155 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
156 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
157 new Exception("The dependency promise is failed", error)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
158 );
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
159 },
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
160 () => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
161 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
162 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
163 new Exception("The dependency promise is cancelled")
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
164 );
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
165 }
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
166 );
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
167 i++;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
168 }
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
169
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
170 return medium;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
171 }
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
172
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
173 #if NET_4_5
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
174
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
175 public static Task<T> GetTask<T>(this IPromise<T> that) {
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
176 Safe.ArgumentNotNull(that, "that");
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
177 var tcs = new TaskCompletionSource<T>();
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
178
104
5f10d54b45df renamed Promise.Last -> Promise.On
cin
parents: 101
diff changeset
179 that.On(tcs.SetResult, tcs.SetException, tcs.SetCanceled);
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
180
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
181 return tcs.Task;
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
182 }
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
183
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
184 #endif
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
185 }
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
186 }
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
187