Mercurial > pub > ImplabNet

comparison Implab/Promise.cs @ 192:f1da3afc3521 release v2.1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Слияние с v2
author cin
date Fri, 22 Apr 2016 13:10:34 +0300
parents 8c0b95069066
children 8200ab154c8a
comparison
equal deleted inserted replaced
71:1714fd8678ef 192:f1da3afc3521
1 using System; 1 using System;
2 using System.Collections.Generic;
3 using System.Reflection;
4 using System.Diagnostics;
5 using System.Threading;
6 using Implab.Parallels; 2 using Implab.Parallels;
7 3
8 namespace Implab { 4 namespace Implab {
5 public class Promise : AbstractPromise, IDeferred {
6 public static readonly Promise SUCCESS;
9 7
10 public delegate void ErrorHandler(Exception e); 8 static Promise() {
11 public delegate T ErrorHandler<out T>(Exception e); 9 SUCCESS = new Promise();
12 public delegate void ResultHandler<in T>(T result); 10 SUCCESS.Resolve();
13 public delegate TNew ResultMapper<in TSrc, out TNew>(TSrc result);
14 public delegate IPromise<TNew> ChainedOperation<in TSrc, TNew>(TSrc result);
15
16 /// <summary>
17 /// Класс для асинхронного получения результатов. Так называемое "обещание".
18 /// </summary>
19 /// <typeparam name="T">Тип получаемого результата</typeparam>
20 /// <remarks>
21 /// <para>Сервис при обращении к его методу дает обещаиние о выполнении операции,
22 /// клиент получив такое обещание может установить ряд обратных вызово для получения
23 /// событий выполнения обещания, тоесть завершения операции и предоставлении результатов.</para>
24 /// <para>
25 /// Обещение может быть как выполнено, так и выполнено с ошибкой. Для подписки на
26 /// данные события клиент должен использовать методы <c>Then</c>.
27 /// </para>
28 /// <para>
29 /// Сервис, в свою очередь, по окончанию выполнения операции (возможно с ошибкой),
30 /// использует методы <c>Resolve</c> либо <c>Reject</c> для оповещения клиетна о
31 /// выполнении обещания.
32 /// </para>
33 /// <para>
34 /// Если сервер успел выполнить обещание еще до того, как клиент на него подписался,
35 /// то в момент подписки клиента будут вызваны соответсвующие события в синхронном
36 /// режиме и клиент будет оповещен в любом случае. Иначе, обработчики добавляются в
37 /// список в порядке подписания и в этом же порядке они будут вызваны при выполнении
38 /// обещания.
39 /// </para>
40 /// <para>
41 /// Обрабатывая результаты обещания можно преобразовывать результаты либо инициировать
42 /// связанные асинхронные операции, которые также возвращают обещания. Для этого следует
43 /// использовать соответствующую форму методе <c>Then</c>.
44 /// </para>
45 /// <para>
46 /// Также хорошим правилом является то, что <c>Resolve</c> и <c>Reject</c> должен вызывать
47 /// только инициатор обещания иначе могут возникнуть противоречия.
48 /// </para>
49 /// </remarks>
50 public class Promise<T> : IPromise<T> {
51
52 protected struct HandlerDescriptor {
53 public ResultHandler<T> resultHandler;
54 public ErrorHandler errorHandler;
55 public Action cancellHandler;
56
57 public void Resolve(T result) {
58 if (resultHandler != null)
59 try {
60 resultHandler(result);
61 } catch (Exception e) {
62 Reject(e);
63 }
64 }
65
66 public void Reject(Exception err) {
67 if (errorHandler != null)
68 try {
69 errorHandler(err);
70 } catch {
71 }
72 }
73
74 public void Cancel() {
75 if (cancellHandler != null)
76 try {
77 cancellHandler();
78 } catch {
79 }
80 }
81 } 11 }
82 12
83 const int UNRESOLVED_SATE = 0; 13 public void Resolve() {
84 const int TRANSITIONAL_STATE = 1; 14 SetResult();
85 const int SUCCEEDED_STATE = 2;
86 const int REJECTED_STATE = 3;
87 const int CANCELLED_STATE = 4;
88
89 readonly bool m_cancellable;
90
91 int m_childrenCount = 0;
92 int m_state;
93 T m_result;
94 Exception m_error;
95
96 readonly MTQueue<HandlerDescriptor> m_handlers = new MTQueue<HandlerDescriptor>();
97
98 public Promise() {
99 m_cancellable = true;
100 } 15 }
101 16
102 public Promise(IPromise parent, bool cancellable) { 17 public void Reject(Exception error) {
103 m_cancellable = cancellable; 18 SetError(error);
104 if (parent != null)
105 AddHandler(
106 null,
107 null,
108 () => {
109 if (parent.IsExclusive)
110 parent.Cancel();
111 }
112 );
113 } 19 }
114
115 bool BeginTransit() {
116 return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE);
117 }
118
119 void CompleteTransit(int state) {
120 if (TRANSITIONAL_STATE != Interlocked.CompareExchange(ref m_state, state, TRANSITIONAL_STATE))
121 throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state");
122 }
123
124 void WaitTransition() {
125 while (m_state == TRANSITIONAL_STATE) {
126 /* noop */
127 }
128 }
129
130 public bool IsResolved {
131 get {
132 return m_state > 1;
133 }
134 }
135
136 public bool IsCancelled {
137 get {
138 return m_state == CANCELLED_STATE;
139 }
140 }
141
142 public Type PromiseType {
143 get { return typeof(T); }
144 }
145
146 /// <summary>
147 /// Выполняет обещание, сообщая об успешном выполнении.
148 /// </summary>
149 /// <param name="result">Результат выполнения.</param>
150 /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
151 public void Resolve(T result) {
152 if (BeginTransit()) {
153 m_result = result;
154 CompleteTransit(SUCCEEDED_STATE);
155 OnStateChanged();
156 } else {
157 WaitTransition();
158 if (m_state != CANCELLED_STATE)
159 throw new InvalidOperationException("The promise is already resolved");
160 }
161 }
162
163 /// <summary>
164 /// Выполняет обещание, сообщая об успешном выполнении. Результатом выполнения будет пустое значения.
165 /// </summary>
166 /// <remarks>
167 /// Данный вариант удобен в случаях, когда интересен факт выполнения операции, нежели полученное значение.
168 /// </remarks>
169 public void Resolve() {
170 Resolve(default(T));
171 }
172
173 /// <summary>
174 /// Выполняет обещание, сообщая об ошибке
175 /// </summary>
176 /// <remarks>
177 /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
178 /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
179 /// будут проигнорированы.
180 /// </remarks>
181 /// <param name="error">Исключение возникшее при выполнении операции</param>
182 /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
183 public void Reject(Exception error) {
184 if (BeginTransit()) {
185 m_error = error;
186 CompleteTransit(REJECTED_STATE);
187 OnStateChanged();
188 } else {
189 WaitTransition();
190 if (m_state == SUCCEEDED_STATE)
191 throw new InvalidOperationException("The promise is already resolved");
192 }
193 }
194
195 /// <summary>
196 /// Отменяет операцию, если это возможно.
197 /// </summary>
198 /// <returns><c>true</c> Операция была отменена, обработчики не будут вызваны.<c>false</c> отмена не возможна, поскольку обещание уже выполнено и обработчики отработали.</returns>
199 public bool Cancel() {
200 if (BeginTransit()) {
201 CompleteTransit(CANCELLED_STATE);
202 OnStateChanged();
203 return true;
204 } else {
205 return false;
206 }
207 }
208
209 // сделано для возвращаемого типа void
210 protected void InternalCancel() {
211 Cancel();
212 }
213
214 /// <summary>
215 /// Adds new handlers to this promise.
216 /// </summary>
217 /// <param name="success">The handler of the successfully completed operation.
218 /// This handler will recieve an operation result as a parameter.</param>
219 /// <param name="error">Handles an exception that may occur during the operation.</param>
220 /// <returns>The new promise chained to this one.</returns>
221 public IPromise<T> Then(ResultHandler<T> success, ErrorHandler error) {
222 if (success == null && error == null)
223 return this;
224
225 var medium = new Promise<T>(this, true);
226
227 ResultHandler<T> resultHandler;
228 if (success != null)
229 resultHandler = x => {
230 success(x);
231 medium.Resolve(x);
232 };
233 else
234 resultHandler = medium.Resolve;
235
236 ErrorHandler errorHandler;
237 if (error != null)
238 errorHandler = x => {
239 // несмотря на то, что обработчик ошибки вызывается безопасно,
240 // т.е. возникшие в нем ошибки будут подавлены, нам нужно
241 // гарантировать, что ошибка будет передана дальше по цепочке обещаний
242 try {
243 error(x);
244 } catch { }
245 medium.Reject(x);
246 };
247 else
248 errorHandler = medium.Reject;
249
250 AddHandler(resultHandler, errorHandler, medium.InternalCancel);
251
252 return medium;
253 }
254
255 public IPromise Then(Action success, ErrorHandler error) {
256 return Then(x => success(), error);
257 }
258
259 public IPromise Then(Action success) {
260 return Then(x => success());
261 }
262
263 /// <summary>
264 /// Adds new handlers to this promise.
265 /// </summary>
266 /// <param name="success">The handler of the successfully completed operation.
267 /// This handler will recieve an operation result as a parameter.</param>
268 /// <param name="error">Handles an exception that may occur during the operation and returns the value which will be used as the result of the operation.</param>
269 /// <returns>The new promise chained to this one.</returns>
270 public IPromise<T> Then(ResultHandler<T> success, ErrorHandler<T> error) {
271 if (success == null && error == null)
272 return this;
273
274 var medium = new Promise<T>(this, true);
275
276 ResultHandler<T> resultHandler;
277 ErrorHandler errorHandler;
278
279 if (success != null)
280 resultHandler = x => {
281 success(x);
282 medium.Resolve(x);
283 };
284 else
285 resultHandler = medium.Resolve;
286
287 if (error != null)
288 errorHandler = x => {
289 try {
290 medium.Resolve(error(x));
291 } catch (Exception e) {
292 medium.Reject(e);
293 }
294 };
295 else
296 errorHandler = medium.Reject;
297
298 AddHandler(resultHandler, errorHandler, medium.InternalCancel);
299
300 return medium;
301 }
302
303
304 public IPromise<T> Then(ResultHandler<T> success) {
305 if (success == null)
306 return this;
307
308 var medium = new Promise<T>(this, true);
309
310 ResultHandler<T> resultHandler;
311
312 if (success != null)
313 resultHandler = x => {
314 success(x);
315 medium.Resolve(x);
316 };
317 else
318 resultHandler = medium.Resolve;
319
320 AddHandler(resultHandler, medium.Reject, medium.InternalCancel);
321
322 return medium;
323 }
324
325 public IPromise<T> Error(ErrorHandler error) {
326 return Then((ResultHandler<T>)null, error);
327 }
328
329 /// <summary>
330 /// Handles error and allows to keep the promise.
331 /// </summary>
332 /// <remarks>
333 /// If the specified handler throws an exception, this exception will be used to reject the promise.
334 /// </remarks>
335 /// <param name="handler">The error handler which returns the result of the promise.</param>
336 /// <returns>New promise.</returns>
337 public IPromise<T> Error(ErrorHandler<T> handler) {
338 if (handler == null)
339 return this;
340
341 var medium = new Promise<T>(this, true);
342
343 AddHandler(
344 x => medium.Resolve(x),
345 e => {
346 try {
347 medium.Resolve(handler(e));
348 } catch (Exception e2) {
349 medium.Reject(e2);
350 }
351 },
352 medium.InternalCancel
353 );
354
355 return medium;
356 }
357
358 public IPromise<T> Anyway(Action handler) {
359 if (handler == null)
360 return this;
361
362 var medium = new Promise<T>(this,true);
363
364 AddHandler(
365 x => {
366 // to avoid handler being called multiple times we handle exception by ourselfs
367 try {
368 handler();
369 medium.Resolve(x);
370 } catch (Exception e) {
371 medium.Reject(e);
372 }
373 },
374
375 e => {
376 try {
377 handler();
378 } catch { }
379 medium.Reject(e);
380 },
381
382 medium.InternalCancel
383 );
384
385 return medium;
386 }
387
388 /// <summary>
389 /// Позволяет преобразовать результат выполения операции к новому типу.
390 /// </summary>
391 /// <typeparam name="TNew">Новый тип результата.</typeparam>
392 /// <param name="mapper">Преобразование результата к новому типу.</param>
393 /// <param name="error">Обработчик ошибки. Данный обработчик получит
394 /// исключение возникшее при выполнении операции.</param>
395 /// <returns>Новое обещание, которое будет выполнено при выполнении исходного обещания.</returns>
396 public IPromise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper, ErrorHandler error) {
397 if (mapper == null)
398 throw new ArgumentNullException("mapper");
399
400 // создаем прицепленное обещание
401 var chained = new Promise<TNew>(this,true);
402
403 ResultHandler<T> resultHandler = result => chained.Resolve(mapper(result));
404 ErrorHandler errorHandler = delegate(Exception e) {
405 if (error != null)
406 try {
407 error(e);
408 } catch { }
409 // в случае ошибки нужно передать исключение дальше по цепочке
410 chained.Reject(e);
411 };
412
413
414 AddHandler(
415 resultHandler,
416 errorHandler,
417 chained.InternalCancel
418 );
419
420 return chained;
421 }
422
423 public IPromise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper) {
424 return Map(mapper, null);
425 }
426
427 /// <summary>
428 /// Сцепляет несколько аснхронных операций. Указанная асинхронная операция будет вызвана после
429 /// выполнения текущей, а результат текущей операции может быть использован для инициализации
430 /// новой операции.
431 /// </summary>
432 /// <typeparam name="TNew">Тип результата указанной асинхронной операции.</typeparam>
433 /// <param name="chained">Асинхронная операция, которая должна будет начаться после выполнения текущей.</param>
434 /// <param name="error">Обработчик ошибки. Данный обработчик получит
435 /// исключение возникшее при выполнении текуещй операции.</param>
436 /// <returns>Новое обещание, которое будет выполнено по окончанию указанной аснхронной операции.</returns>
437 public IPromise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained, ErrorHandler error) {
438
439 // проблема в том, что на момент связывания еще не начата асинхронная операция, поэтому нужно
440 // создать посредника, к которому будут подвызяваться следующие обработчики.
441 // когда будет выполнена реальная асинхронная операция, она обратиться к посреднику, чтобы
442 // передать через него результаты работы.
443 var medium = new Promise<TNew>(this, true);
444
445 ResultHandler<T> resultHandler = delegate(T result) {
446 if (medium.IsCancelled)
447 return;
448
449 var promise = chained(result);
450
451 promise.Then(
452 x => medium.Resolve(x),
453 e => medium.Reject(e)
454 );
455
456 // notify chained operation that it's not needed anymore
457 // порядок вызова Then, Cancelled важен, поскольку от этого
458 // зависит IsExclusive
459 medium.Cancelled(() => {
460 if(promise.IsExclusive)
461 promise.Cancel();
462 });
463
464 // внешняя отмена связанной операции рассматривается как ошибка
465 promise.Cancelled(() => medium.Reject(new OperationCanceledException()));
466 };
467
468 ErrorHandler errorHandler = delegate(Exception e) {
469 if (error != null)
470 error(e);
471 // в случае ошибки нужно передать исключение дальше по цепочке
472 medium.Reject(e);
473 };
474
475 AddHandler(
476 resultHandler,
477 errorHandler,
478 medium.InternalCancel
479 );
480
481 return medium;
482 }
483
484 public IPromise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
485 return Chain(chained, null);
486 }
487
488 public IPromise<T> Cancelled(Action handler) {
489 AddHandler(null, null, handler);
490 return this;
491 }
492
493 /// <summary>
494 /// Adds the specified handler for all cases (success, error, cancel)
495 /// </summary>
496 /// <param name="handler">The handler that will be called anyway</param>
497 /// <returns>self</returns>
498 public IPromise<T> Finally(Action handler) {
499 if (handler == null)
500 throw new ArgumentNullException("handler");
501 AddHandler(
502 x => handler(),
503 e => handler(),
504 handler
505 );
506 return this;
507 }
508
509 /// <summary>
510 /// Преобразует результат обещания к нужному типу
511 /// </summary>
512 /// <typeparam name="T2"></typeparam>
513 /// <returns></returns>
514 public IPromise<T2> Cast<T2>() {
515 return Map(x => (T2)(object)x, null);
516 }
517
518 /// <summary>
519 /// Дожидается отложенного обещания и в случае успеха, возвращает
520 /// его, результат, в противном случае бросает исключение.
521 /// </summary>
522 /// <remarks>
523 /// <para>
524 /// Если ожидание обещания было прервано по таймауту, это не значит,
525 /// что обещание было отменено или что-то в этом роде, это только
526 /// означает, что мы его не дождались, однако все зарегистрированные
527 /// обработчики, как были так остались и они будут вызваны, когда
528 /// обещание будет выполнено.
529 /// </para>
530 /// <para>
531 /// Такое поведение вполне оправдано поскольку таймаут может истечь
532 /// в тот момент, когда началась обработка цепочки обработчиков, и
533 /// к тому же текущее обещание может стоять в цепочке обещаний и его
534 /// отклонение может привести к непрогнозируемому результату.
535 /// </para>
536 /// </remarks>
537 /// <param name="timeout">Время ожидания</param>
538 /// <returns>Результат выполнения обещания</returns>
539 public T Join(int timeout) {
540 var evt = new ManualResetEvent(false);
541 Anyway(() => evt.Set());
542 Cancelled(() => evt.Set());
543
544 if (!evt.WaitOne(timeout, true))
545 throw new TimeoutException();
546
547 switch (m_state) {
548 case SUCCEEDED_STATE:
549 return m_result;
550 case CANCELLED_STATE:
551 throw new OperationCanceledException();
552 case REJECTED_STATE:
553 throw new TargetInvocationException(m_error);
554 default:
555 throw new ApplicationException(String.Format("Invalid promise state {0}", m_state));
556 }
557 }
558
559 public T Join() {
560 return Join(Timeout.Infinite);
561 }
562
563 void AddHandler(ResultHandler<T> success, ErrorHandler error, Action cancel) {
564 if (success != null || error != null)
565 Interlocked.Increment(ref m_childrenCount);
566
567 HandlerDescriptor handler = new HandlerDescriptor {
568 resultHandler = success,
569 errorHandler = error,
570 cancellHandler = cancel
571 };
572
573 bool queued;
574
575 if (!IsResolved) {
576 m_handlers.Enqueue(handler);
577 queued = true;
578 } else {
579 // the promise is in resolved state, just invoke the handled with minimum overhead
580 queued = false;
581 InvokeHandler(handler);
582 }
583
584 if (queued && IsResolved && m_handlers.TryDequeue(out handler))
585 // if the promise have been resolved while we was adding handler to the queue
586 // we can't guarantee that someone is still processing it
587 // therefore we will fetch a handler from the queue and execute it
588 // note that fetched handler may be not the one that we have added
589 // even we can fetch no handlers at all :)
590 InvokeHandler(handler);
591 }
592
593 protected virtual void InvokeHandler(HandlerDescriptor handler) {
594 switch (m_state) {
595 case SUCCEEDED_STATE:
596 handler.Resolve(m_result);
597 break;
598 case REJECTED_STATE:
599 handler.Reject(m_error);
600 break;
601 case CANCELLED_STATE:
602 handler.Cancel();
603 break;
604 default:
605 // do nothing
606 return;
607 }
608 }
609
610 void OnStateChanged() {
611 HandlerDescriptor handler;
612 while (m_handlers.TryDequeue(out handler))
613 InvokeHandler(handler);
614 }
615
616 public bool IsExclusive {
617 get {
618 return m_childrenCount <= 1;
619 }
620 }
621
622 /// <summary>
623 /// Объединяет несколько обещаний в одно, результатом которого является массив результатов других обещаний.
624 /// Если хотябы одно из переданных обещаний не будет выполнено, то новое обещение тоже не будет выполнено.
625 /// При отмене нового обещания, переданные обещания также будут отменены, если никто больше на них не подписан.
626 /// </summary>
627 /// <param name="promises">Список обещаний. Если список пустой, то результирующее обещание возвращается уже выполненным.</param>
628 /// <returns>Обещание объединяющее в себе результат переданных обещаний.</returns>
629 /// <exception cref="ArgumentNullException"><paramref name="promises"/> не может быть null</exception>
630 public static IPromise<T[]> CreateComposite(IList<IPromise<T>> promises) {
631 if (promises == null)
632 throw new ArgumentNullException();
633
634 // создаем аккумулятор для результатов и результирующее обещание
635 var result = new T[promises.Count];
636 var promise = new Promise<T[]>();
637
638 // special case
639 if (promises.Count == 0) {
640 promise.Resolve(result);
641 return promise;
642 }
643
644 int pending = promises.Count;
645
646 for (int i = 0; i < promises.Count; i++) {
647 var dest = i;
648
649 if (promises[i] != null) {
650 promises[i].Then(
651 x => {
652 result[dest] = x;
653 if (Interlocked.Decrement(ref pending) == 0)
654 promise.Resolve(result);
655 },
656 e => promise.Reject(e)
657 );
658 } else {
659 if (Interlocked.Decrement(ref pending) == 0)
660 promise.Resolve(result);
661 }
662 }
663
664 promise.Cancelled(
665 () => {
666 foreach (var d in promises)
667 if (d != null && d.IsExclusive)
668 d.Cancel();
669 }
670 );
671
672 return promise;
673 }
674
675 /// <summary>
676 /// Объединяет несколько обещаний в одно. Результирующее обещание будет выполнено при
677 /// выполнении всех указанных обещаний. При этом возвращаемые значения первичных обещаний
678 /// игнорируются.
679 /// </summary>
680 /// <param name="promises">Коллекция первичных обещаний, которые будут объеденены в одно.</param>
681 /// <returns>Новое обещание, объединяющее в себе переданные.</returns>
682 /// <remarks>
683 /// Если в коллекции встречаюься <c>null</c>, то они воспринимаются как выполненные обещания.
684 /// </remarks>
685 public static IPromise CreateComposite(ICollection<IPromise> promises) {
686 if (promises == null)
687 throw new ArgumentNullException();
688 if (promises.Count == 0)
689 return Promise<object>.ResultToPromise(null);
690
691 int countdown = promises.Count;
692
693 var result = new Promise<object>();
694
695 foreach (var d in promises) {
696 if (d == null) {
697 if (Interlocked.Decrement(ref countdown) == 0)
698 result.Resolve(null);
699 } else {
700 d.Then(() => {
701 if (Interlocked.Decrement(ref countdown) == 0)
702 result.Resolve(null);
703 });
704 }
705 }
706
707 result.Cancelled(() => {
708 foreach (var d in promises)
709 if (d != null && d.IsExclusive)
710 d.Cancel();
711 });
712
713 return result;
714 }
715
716 public static Promise<T> ResultToPromise(T result) {
717 var p = new Promise<T>();
718 p.Resolve(result);
719 return p;
720 }
721
722 public static Promise<T> ExceptionToPromise(Exception error) {
723 if (error == null)
724 throw new ArgumentNullException();
725
726 var p = new Promise<T>();
727 p.Reject(error);
728 return p;
729 }
730
731 #region IPromiseBase explicit implementation
732
733 IPromise IPromise.Error(ErrorHandler error) {
734 return Error(error);
735 }
736
737 IPromise IPromise.Anyway(Action handler) {
738 return Anyway(handler);
739 }
740
741 IPromise IPromise.Finally(Action handler) {
742 return Finally(handler);
743 }
744
745 IPromise IPromise.Cancelled(Action handler) {
746 return Cancelled(handler);
747 }
748
749 void IPromise.Join() {
750 Join();
751 }
752
753 void IPromise.Join(int timeout) {
754 Join(timeout);
755 }
756
757 #endregion
758
759
760
761 } 20 }
762 } 21 }
22