Mercurial > pub > ImplabNet
annotate Implab/PromiseExtensions.cs @ 160:5802131432e4 v2
fixed regression: race condition in Promise
DFA refactoring
| author | cin |
|---|---|
| date | Thu, 18 Feb 2016 19:38:54 +0300 |
| parents | ec91a6dfa5b3 |
| children | 822aab37b107 |
| rev | line source |
|---|---|
| 72 | 1 using System.Threading; |
| 75 | 2 using System; |
| 109 | 3 using Implab.Diagnostics; |
|
119
2573b562e328
Promises rewritten, added improved version of AsyncQueue
cin
parents:
110
diff
changeset
|
4 using System.Collections.Generic; |
| 109 | 5 |
| 6 | |
| 75 | 7 #if NET_4_5 |
| 8 using System.Threading.Tasks; | |
| 9 #endif | |
| 72 | 10 |
| 11 namespace Implab { | |
| 12 public static class PromiseExtensions { | |
| 13 public static IPromise<T> DispatchToCurrentContext<T>(this IPromise<T> that) { | |
| 75 | 14 Safe.ArgumentNotNull(that, "that"); |
| 72 | 15 var context = SynchronizationContext.Current; |
| 16 if (context == null) | |
| 17 return that; | |
| 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 | 21 |
| 104 | 22 that.On( |
| 76 | 23 p.Resolve, |
| 24 p.Reject, | |
| 25 p.Cancel | |
| 72 | 26 ); |
| 27 return p; | |
| 28 } | |
| 29 | |
| 30 public static IPromise<T> DispatchToContext<T>(this IPromise<T> that, SynchronizationContext context) { | |
| 75 | 31 Safe.ArgumentNotNull(that, "that"); |
| 72 | 32 Safe.ArgumentNotNull(context, "context"); |
| 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 | 37 |
| 104 | 38 that.On( |
| 76 | 39 p.Resolve, |
| 40 p.Reject, | |
| 41 p.Cancel | |
| 72 | 42 ); |
| 43 return p; | |
| 44 } | |
| 75 | 45 |
| 101 | 46 /// <summary> |
| 47 /// Ensures the dispatched. | |
| 48 /// </summary> | |
| 49 /// <returns>The dispatched.</returns> | |
| 50 /// <param name="that">That.</param> | |
| 51 /// <param name="head">Head.</param> | |
| 52 /// <param name="cleanup">Cleanup.</param> | |
| 53 /// <typeparam name="TPromise">The 1st type parameter.</typeparam> | |
| 54 /// <typeparam name="T">The 2nd type parameter.</typeparam> | |
| 55 public static TPromise EnsureDispatched<TPromise,T>(this TPromise that, IPromise<T> head, Action<T> cleanup) where TPromise : IPromise{ | |
| 56 Safe.ArgumentNotNull(that, "that"); | |
| 57 Safe.ArgumentNotNull(head, "head"); | |
| 58 | |
|
138
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
59 that.On(() => head.On(cleanup), PromiseEventType.Cancelled); |
| 101 | 60 |
| 61 return that; | |
| 62 } | |
| 63 | |
| 75 | 64 public static AsyncCallback AsyncCallback<T>(this Promise<T> that, Func<IAsyncResult,T> callback) { |
| 65 Safe.ArgumentNotNull(that, "that"); | |
| 66 Safe.ArgumentNotNull(callback, "callback"); | |
| 109 | 67 var op = TraceContext.Instance.CurrentOperation; |
| 75 | 68 return ar => { |
| 109 | 69 TraceContext.Instance.EnterLogicalOperation(op,false); |
| 75 | 70 try { |
| 71 that.Resolve(callback(ar)); | |
| 72 } catch (Exception err) { | |
| 73 that.Reject(err); | |
| 109 | 74 } finally { |
| 75 TraceContext.Instance.Leave(); | |
| 75 | 76 } |
| 77 }; | |
| 78 } | |
| 110 | 79 |
| 80 static void CancelCallback(object cookie) { | |
| 81 ((ICancellable)cookie).Cancel(); | |
| 82 } | |
| 83 | |
| 84 /// <summary> | |
| 85 /// Cancells promise after the specified timeout is elapsed. | |
| 86 /// </summary> | |
| 87 /// <param name="that">The promise to cancel on timeout.</param> | |
| 88 /// <param name="milliseconds">The timeout in milliseconds.</param> | |
| 89 /// <typeparam name="TPromise">The 1st type parameter.</typeparam> | |
| 90 public static TPromise Timeout<TPromise>(this TPromise that, int milliseconds) where TPromise : IPromise { | |
| 91 Safe.ArgumentNotNull(that, "that"); | |
| 92 var timer = new Timer(CancelCallback, that, milliseconds, -1); | |
| 93 that.On(timer.Dispose, PromiseEventType.All); | |
| 94 return that; | |
| 95 } | |
|
119
2573b562e328
Promises rewritten, added improved version of AsyncQueue
cin
parents:
110
diff
changeset
|
96 |
| 124 | 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 | 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 | 104 if (count == 0) { |
| 105 medium.Resolve(); | |
| 106 return medium; | |
| 107 } | |
| 108 | |
| 124 | 109 medium.On(() => { |
| 110 foreach(var p2 in that) | |
| 111 p2.Cancel(); | |
| 112 }, PromiseEventType.ErrorOrCancel); | |
| 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 | 121 if (Interlocked.Increment(ref errors) == 1) |
| 122 medium.Reject( | |
| 123 new Exception("The dependency promise is failed", error) | |
| 124 ); | |
|
119
2573b562e328
Promises rewritten, added improved version of AsyncQueue
cin
parents:
110
diff
changeset
|
125 }, |
|
138
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
126 reason => { |
| 124 | 127 if (Interlocked.Increment(ref errors) == 1) |
|
138
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
128 medium.Cancel( |
| 124 | 129 new Exception("The dependency promise is cancelled") |
| 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 | 136 |
| 137 public static IPromise<T[]> Bundle<T>(this ICollection<IPromise<T>> that) { | |
| 138 Safe.ArgumentNotNull(that, "that"); | |
| 139 | |
| 140 int count = that.Count; | |
| 141 int errors = 0; | |
| 142 var medium = new Promise<T[]>(); | |
| 143 var results = new T[that.Count]; | |
| 144 | |
| 145 medium.On(() => { | |
| 146 foreach(var p2 in that) | |
| 147 p2.Cancel(); | |
| 148 }, PromiseEventType.ErrorOrCancel); | |
| 149 | |
| 150 int i = 0; | |
| 151 foreach (var p in that) { | |
| 152 var idx = i; | |
| 153 p.On( | |
| 154 x => { | |
| 155 results[idx] = x; | |
| 156 if (Interlocked.Decrement(ref count) == 0) | |
| 157 medium.Resolve(results); | |
| 158 }, | |
| 159 error => { | |
| 160 if (Interlocked.Increment(ref errors) == 1) | |
| 161 medium.Reject( | |
| 162 new Exception("The dependency promise is failed", error) | |
| 163 ); | |
| 164 }, | |
|
138
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
165 reason => { |
| 124 | 166 if (Interlocked.Increment(ref errors) == 1) |
|
138
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
167 medium.Cancel( |
|
f75cfa58e3d4
added ICancellable.Cancel(Exception) to allow specify the reason of cancellation
cin
parents:
136
diff
changeset
|
168 new Exception("The dependency promise is cancelled", reason) |
| 124 | 169 ); |
| 170 } | |
| 171 ); | |
| 172 i++; | |
| 173 } | |
| 174 | |
| 175 return medium; | |
| 176 } | |
| 145 | 177 |
| 178 public static IPromise Then(this IPromise that, Action success, Action<Exception> error, Action<Exception> cancel) { | |
| 179 Safe.ArgumentNotNull(that, "that"); | |
| 180 | |
| 149 | 181 var d = new ActionTask(success, error, cancel, false); |
| 145 | 182 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 183 if (success != null) | |
| 184 d.CancellationRequested(that.Cancel); | |
| 185 return d; | |
| 186 } | |
| 187 | |
| 188 public static IPromise Then(this IPromise that, Action success, Action<Exception> error) { | |
| 189 return Then(that, success, error, null); | |
| 190 } | |
| 191 | |
| 192 public static IPromise Then(this IPromise that, Action success) { | |
| 193 return Then(that, success, null, null); | |
| 194 } | |
| 195 | |
| 196 public static IPromise<T> Then<T>(this IPromise that, Func<T> success, Func<Exception, T> error, Func<Exception, T> cancel) { | |
| 197 Safe.ArgumentNotNull(that, "that"); | |
| 198 | |
| 149 | 199 var d = new FuncTask<T>(success, error, cancel, false); |
| 145 | 200 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 201 if (success != null) | |
| 202 d.CancellationRequested(that.Cancel); | |
| 203 return d; | |
| 204 } | |
| 205 | |
| 206 public static IPromise<T> Then<T>(this IPromise that, Func<T> success, Func<Exception, T> error) { | |
| 207 return Then(that, success, error, null); | |
| 208 } | |
| 209 | |
| 210 public static IPromise<T> Then<T>(this IPromise that, Func<T> success) { | |
| 211 return Then(that, success, null, null); | |
| 212 } | |
| 213 | |
| 214 public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success, Func<Exception, T2> error, Func<Exception, T2> cancel) { | |
| 215 Safe.ArgumentNotNull(that, "that"); | |
| 149 | 216 var d = new FuncTask<T,T2>(success, error, cancel, false); |
| 145 | 217 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 218 if (success != null) | |
| 219 d.CancellationRequested(that.Cancel); | |
| 220 return d; | |
| 221 } | |
| 222 | |
| 223 public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success, Func<Exception, T2> error) { | |
| 224 return Then(that, success, error, null); | |
| 225 } | |
| 226 | |
| 227 public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success) { | |
| 228 return Then(that, success, null, null); | |
| 229 } | |
| 230 | |
| 231 #region chain traits | |
| 232 public static IPromise Chain(this IPromise that, Func<IPromise> success, Func<Exception,IPromise> error, Func<Exception,IPromise> cancel) { | |
| 233 Safe.ArgumentNotNull(that, "that"); | |
| 234 | |
| 149 | 235 var d = new ActionChainTask(success, error, cancel, false); |
| 145 | 236 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 237 if (success != null) | |
| 238 d.CancellationRequested(that.Cancel); | |
| 239 return d; | |
| 240 } | |
| 241 | |
| 242 public static IPromise Chain(this IPromise that, Func<IPromise> success, Func<Exception,IPromise> error) { | |
| 243 return Chain(that, success, error, null); | |
| 244 } | |
| 245 | |
| 246 public static IPromise Chain(this IPromise that, Func<IPromise> success) { | |
| 247 return Chain(that, success, null, null); | |
| 248 } | |
| 249 | |
| 250 public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success, Func<Exception, IPromise<T>> error, Func<Exception, IPromise<T>> cancel) { | |
| 251 Safe.ArgumentNotNull(that, "that"); | |
| 252 | |
| 149 | 253 var d = new FuncChainTask<T>(success, error, cancel, false); |
| 145 | 254 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 255 if (success != null) | |
| 256 d.CancellationRequested(that.Cancel); | |
| 257 return d; | |
| 258 } | |
| 259 | |
| 260 public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success, Func<Exception, IPromise<T>> error) { | |
| 261 return Chain(that, success, error, null); | |
| 262 } | |
| 263 | |
| 264 public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success) { | |
| 265 return Chain(that, success, null, null); | |
| 266 } | |
| 267 | |
| 268 public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success, Func<Exception, IPromise<T2>> error, Func<Exception, IPromise<T2>> cancel) { | |
| 269 Safe.ArgumentNotNull(that, "that"); | |
| 149 | 270 var d = new FuncChainTask<T,T2>(success, error, cancel, false); |
| 145 | 271 that.On(d.Resolve, d.Reject, d.CancelOperation); |
| 272 if (success != null) | |
| 273 d.CancellationRequested(that.Cancel); | |
| 274 return d; | |
| 275 } | |
| 276 | |
| 277 public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success, Func<Exception, IPromise<T2>> error) { | |
| 278 return Chain(that, success, error, null); | |
| 279 } | |
| 280 | |
| 281 public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success) { | |
| 282 return Chain(that, success, null, null); | |
| 283 } | |
| 284 | |
| 285 #endregion | |
| 286 | |
| 75 | 287 |
| 288 #if NET_4_5 | |
| 289 | |
| 151 | 290 public static PromiseAwaiter<T> GetAwaiter<T>(this IPromise<T> that) { |
| 75 | 291 Safe.ArgumentNotNull(that, "that"); |
| 292 | |
| 151 | 293 return new PromiseAwaiter<T>(that); |
| 75 | 294 } |
| 295 | |
| 296 #endif | |
| 72 | 297 } |
| 298 } | |
| 299 |