Mercurial > pub > ImplabNet

annotate Implab/PromiseExtensions.cs @ 137:238e15580926 v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
added the blocking queue
author cin
date Mon, 16 Feb 2015 17:48:39 +0300
parents e9e7940c7d98
children f75cfa58e3d4
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
136
e9e7940c7d98 shared locks + tests
cin
parents: 124
diff changeset
104 if (count == 0) {
e9e7940c7d98 shared locks + tests
cin
parents: 124
diff changeset
105 medium.Resolve();
e9e7940c7d98 shared locks + tests
cin
parents: 124
diff changeset
106 return medium;
e9e7940c7d98 shared locks + tests
cin
parents: 124
diff changeset
107 }
e9e7940c7d98 shared locks + tests
cin
parents: 124
diff changeset
108
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
109 medium.On(() => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
110 foreach(var p2 in that)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
111 p2.Cancel();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
112 }, PromiseEventType.ErrorOrCancel);
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
113
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
114 foreach (var p in that)
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
115 p.On(
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
116 () => {
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
117 if (Interlocked.Decrement(ref count) == 0)
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
118 medium.Resolve();
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
119 },
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
120 error => {
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
121 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
122 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
123 new Exception("The dependency promise is failed", error)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
124 );
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
125 },
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
126 () => {
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
127 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
128 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
129 new Exception("The dependency promise is cancelled")
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
130 );
119
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
131 }
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
132 );
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
133
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
134 return medium;
2573b562e328 Promises rewritten, added improved version of AsyncQueue
cin
parents: 110
diff changeset
135 }
124
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
136
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
137 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
138 Safe.ArgumentNotNull(that, "that");
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 int count = that.Count;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
141 int errors = 0;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
142 var medium = new Promise<T[]>();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
143 var results = new T[that.Count];
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 medium.On(() => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
146 foreach(var p2 in that)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
147 p2.Cancel();
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
148 }, PromiseEventType.ErrorOrCancel);
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
149
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
150 int i = 0;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
151 foreach (var p in that) {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
152 var idx = i;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
153 p.On(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
154 x => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
155 results[idx] = x;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
156 if (Interlocked.Decrement(ref count) == 0)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
157 medium.Resolve(results);
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 error => {
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
160 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
161 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
162 new Exception("The dependency promise is failed", error)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
163 );
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 if (Interlocked.Increment(ref errors) == 1)
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
167 medium.Reject(
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
168 new Exception("The dependency promise is cancelled")
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 }
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
171 );
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
172 i++;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
173 }
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
174
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
175 return medium;
a336cb13c6a9 major update, added Drain mathod to AsyncQueue class
cin
parents: 119
diff changeset
176 }
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
177
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
178 #if NET_4_5
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
179
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
180 public static Task<T> GetTask<T>(this IPromise<T> that) {
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
181 Safe.ArgumentNotNull(that, "that");
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
182 var tcs = new TaskCompletionSource<T>();
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
183
104
5f10d54b45df renamed Promise.Last -> Promise.On
cin
parents: 101
diff changeset
184 that.On(tcs.SetResult, tcs.SetException, tcs.SetCanceled);
75
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
185
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
186 return tcs.Task;
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
187 }
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
188
4439140706d0 major refactoring, added tasks support
cin
parents: 72
diff changeset
189 #endif
72
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
190 }
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
191 }
d67b95eddaf4 promises refactoring
cin
parents:
diff changeset
192