implab/ImplabNet Commit - r28:81708975d6f7

fixes

cin -

r28:81708975d6f7 default

parent child

Context file:

r28:81708975d6f7

Collapse all files

Implab/Promise.cs +1 -1

              using System;
              using System.Collections.Generic;
              using System.Reflection;
              using System.Diagnostics;
              using System.Threading;
              using Implab.Parallels;
              namespace Implab {
                  public delegate void ErrorHandler(Exception e);
                  public delegate T ErrorHandler<out T>(Exception e);
                  public delegate void ResultHandler<in T>(T result);
                  public delegate TNew ResultMapper<in TSrc, out TNew>(TSrc result);
                  public delegate Promise<TNew> ChainedOperation<in TSrc, TNew>(TSrc result);
                  /// <summary>
                  /// Класс для асинхронного получения результатов. Так называемое "обещание".
                  /// </summary>
                  /// <typeparam name="T">Тип получаемого результата</typeparam>
                  /// <remarks>
                  /// <para>Сервис при обращении к его методу дает обещаиние о выполнении операции,
                  /// клиент получив такое обещание может установить ряд обратных вызово для получения
                  /// событий выполнения обещания, тоесть завершения операции и предоставлении результатов.</para>
                  /// <para>
                  /// Обещение может быть как выполнено, так и выполнено с ошибкой. Для подписки на
                  /// данные события клиент должен использовать методы <c>Then</c>.
                  /// </para>
                  /// <para>
                  /// Сервис, в свою очередь, по окончанию выполнения операции (возможно с ошибкой),
                  /// использует методы <c>Resolve</c> либо <c>Reject</c> для оповещения клиетна о
                  /// выполнении обещания.
                  /// </para>
                  /// <para>
                  /// Если сервер успел выполнить обещание еще до того, как клиент на него подписался,
                  /// то в момент подписки клиента будут вызваны соответсвующие события в синхронном
                  /// режиме и клиент будет оповещен в любом случае. Иначе, обработчики добавляются в
                  /// список в порядке подписания и в этом же порядке они будут вызваны при выполнении
                  /// обещания.
                  /// </para>
                  /// <para>
                  /// Обрабатывая результаты обещания можно преобразовывать результаты либо инициировать
                  /// связанные асинхронные операции, которые также возвращают обещания. Для этого следует
                  /// использовать соответствующую форму методе <c>Then</c>.
                  /// </para>
                  /// <para>
                  /// Также хорошим правилом является то, что <c>Resolve</c> и <c>Reject</c> должен вызывать
                  /// только инициатор обещания иначе могут возникнуть противоречия.
                  /// </para>
                  /// </remarks>
                  public class Promise<T> : IPromise<T> {
-                     struct HandlerDescriptor {
+                     protected struct HandlerDescriptor {
                          public ResultHandler<T> resultHandler;
                          public ErrorHandler errorHandler;
                          public Action cancellHandler;
                          public void Resolve(T result) {
                              if (resultHandler != null)
                                  try {
                                      resultHandler(result);
                                  } catch (Exception e) {
                                      Reject(e);
                                  }
                          }
                          public void Reject(Exception err) {
                              if (errorHandler != null)
                                  try {
                                      errorHandler(err);
                                  } catch {
                                  }
                          }
                          public void Cancel() {
                              if (cancellHandler != null)
                                  try {
                                      cancellHandler();
                                  } catch {
                                  }
                          }
                      }
                      const int UnresolvedSate = 0;
                      const int TransitionalState = 1;
                      const int SucceededState = 2;
                      const int RejectedState = 3;
                      const int CancelledState = 4;
                      readonly IPromiseBase m_parent;
                      readonly bool m_cancellable;
                      int m_childrenCount = 0;
                      int m_state;
                      T m_result;
                      Exception m_error;
                      readonly MTQueue<HandlerDescriptor> m_handlers = new MTQueue<HandlerDescriptor>();
                      public Promise() {
                          m_cancellable = true;
                      }
                      public Promise(IPromiseBase parent, bool cancellable) {
                          m_cancellable = cancellable;
                          m_parent = parent;
                      }
                      void InternalCancel() {
                          // don't try to cancel parent :)
                          Cancel(false);
                      }
                      bool BeginTransit() {
                          return UnresolvedSate == Interlocked.CompareExchange(ref m_state, TransitionalState, UnresolvedSate);
                      }
                      void CompleteTransit(int state) {
                          if (TransitionalState != Interlocked.CompareExchange(ref m_state, state, TransitionalState))
                              throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state");
                      }
                      void WaitTransition() {
                          while (m_state == TransitionalState) {
                              /* noop */
                          }
                      }
                      public bool IsResolved {
                          get {
                              return m_state > 1;
                          }
                      }
                      public bool IsCancelled {
                          get {
                              return m_state == CancelledState;
                          }
                      }
                      /// <summary>
                      /// Выполняет обещание, сообщая об успешном выполнении.
                      /// </summary>
                      /// <param name="result">Результат выполнения.</param>
                      /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                      public void Resolve(T result) {
                          if (BeginTransit()) {
                              m_result = result;
                              CompleteTransit(SucceededState);
                              OnStateChanged();
                          } else {
                              WaitTransition();
                              if (m_state != CancelledState)
                                  throw new InvalidOperationException("The promise is already resolved");
                          }
                      }
                      /// <summary>
                      /// Выполняет обещание, сообщая об ошибке
                      /// </summary>
                      /// <remarks>
                      /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                      /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                      /// будут проигнорированы.
                      /// </remarks>
                      /// <param name="error">Исключение возникшее при выполнении операции</param>
                      /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                      public void Reject(Exception error) {
                          if (BeginTransit()) {
                              m_error = error;
                              CompleteTransit(RejectedState);
                              OnStateChanged();
                          } else {
                              WaitTransition();
                              if (m_state == SucceededState)
                                  throw new InvalidOperationException("The promise is already resolved");
                          }
                      }
                      /// <summary>
                      /// Отменяет операцию, если это возможно.
                      /// </summary>
                      /// <returns><c>true</c> Операция была отменена, обработчики не будут вызваны.<c>false</c> отмена не возможна, поскольку обещание уже выполнено и обработчики отработали.</returns>
                      public bool Cancel() {
                          return Cancel(true);
                      }
                      /// <summary>
                      /// Adds new handlers to this promise.
                      /// </summary>
                      /// <param name="success">The handler of the successfully completed operation.
                      /// This handler will recieve an operation result as a parameter.</param>
                      /// <param name="error">Handles an exception that may occur during the operation.</param>
                      /// <returns>The new promise chained to this one.</returns>
                      public IPromise<T> Then(ResultHandler<T> success, ErrorHandler error) {
                          if (success == null && error == null)
                              return this;
                          var medium = new Promise<T>(this, true);
                          ResultHandler<T> resultHandler;
                          if (success != null)
                              resultHandler = x => {
                                  success(x);
                                  medium.Resolve(x);
                              };
                          else
                              resultHandler = medium.Resolve;
                          ErrorHandler errorHandler;
                          if (error != null)
                              errorHandler = x => {
                                  // несмотря на то, что обработчик ошибки вызывается безопасно,
                                  // т.е. возникшие в нем ошибки будут подавлены, нам нужно
                                  // гарантировать, что ошибка будет передана дальше по цепочке обещаний
                                  try {
                                      error(x);
                                  } catch { }
                                  medium.Reject(x);
                              };
                          else
                              errorHandler = medium.Reject;
                          AddHandler(resultHandler, errorHandler, medium.InternalCancel);
                          return medium;
                      }
                      public IPromiseBase Then(Action success,ErrorHandler error)
                      {
                          return Then(x => success(), error);
                      }
                      public IPromiseBase Then(Action success)
                      {
                          return Then(success);
                      }
                      /// <summary>
                      /// Adds new handlers to this promise.
                      /// </summary>
                      /// <param name="success">The handler of the successfully completed operation.
                      /// This handler will recieve an operation result as a parameter.</param>
                      /// <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>
                      /// <returns>The new promise chained to this one.</returns>
                      public IPromise<T> Then(ResultHandler<T> success, ErrorHandler<T> error) {
                          if (success == null && error == null)
                              return this;
                          var medium = new Promise<T>(this, true);
                          ResultHandler<T> resultHandler;
                          ErrorHandler errorHandler;
                          if (success != null)
                              resultHandler = x => {
                                  success(x);
                                  medium.Resolve(x);
                              };
                          else
                              resultHandler = medium.Resolve;
                          if (error != null)
                              errorHandler = x => {
                                  try {
                                      medium.Resolve(error(x));
                                  } catch(Exception e) {
                                      medium.Reject(e);
                                  }
                              };
                          else
                              errorHandler = medium.Reject;
                          AddHandler(resultHandler, errorHandler, medium.InternalCancel);
                          return medium;
                      }
                      public IPromise<T> Then(ResultHandler<T> success) {
                          if (success == null)
                              return this;
                          var medium = new Promise<T>(this, true);
                          ResultHandler<T> resultHandler;
                          if (success != null)
                              resultHandler = x => {
                                  success(x);
                                  medium.Resolve(x);
                              };
                          else
                              resultHandler = medium.Resolve;
                          AddHandler(resultHandler, medium.Reject, medium.InternalCancel);
                          return medium;
                      }
                      public IPromise<T> Error(ErrorHandler error) {
                          return Then((ResultHandler<T>)null, error);
                      }
                      /// <summary>
                      /// Handles error and allows to keep the promise.
                      /// </summary>
                      /// <remarks>
                      /// If the specified handler throws an exception, this exception will be used to reject the promise.
                      /// </remarks>
                      /// <param name="handler">The error handler which returns the result of the promise.</param>
                      /// <returns>New promise.</returns>
                      public IPromise<T> Error(ErrorHandler<T> handler) {
                          if (handler == null)
                              return this;
                          var medium = new Promise<T>(this, true);
                          AddHandler(
                              x => medium.Resolve(x),
                              e => {
                                  try {
                                      medium.Resolve(handler(e));
                                  } catch (Exception e2) {
                                      medium.Reject(e2);
                                  }
                              },
                              medium.InternalCancel
                          );
                          return medium;
                      }
                      public IPromise<T> Anyway(Action handler) {
                          if (handler == null)
                              return this;
                          var medium = new Promise<T>();
                          AddHandler(
                              x => {
                                  // to avoid handler being called multiple times we handle exception by ourselfs
                                  try {
                                      handler();
                                      medium.Resolve(x);
                                  } catch (Exception e) {
                                      medium.Reject(e);
                                  }
                              },
                              e => {
                                  try {
                                      handler();
                                  } catch { }
                                  medium.Reject(e);
                              },
                              medium.InternalCancel
                          );
                          return medium;
                      }
                      /// <summary>
                      /// Позволяет преобразовать результат выполения операции к новому типу.
                      /// </summary>
                      /// <typeparam name="TNew">Новый тип результата.</typeparam>
                      /// <param name="mapper">Преобразование результата к новому типу.</param>
                      /// <param name="error">Обработчик ошибки. Данный обработчик получит
                      /// исключение возникшее при выполнении операции.</param>
                      /// <returns>Новое обещание, которое будет выполнено при выполнении исходного обещания.</returns>
                      public IPromise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper, ErrorHandler error) {
                          if (mapper == null)
                              throw new ArgumentNullException("mapper");
                          // создаем прицепленное обещание
                          var chained = new Promise<TNew>();
                          ResultHandler<T> resultHandler = result => chained.Resolve(mapper(result));
                          ErrorHandler errorHandler = delegate(Exception e) {
                              if (error != null)
                                  try {
                                      error(e);
                                  } catch { }
                              // в случае ошибки нужно передать исключение дальше по цепочке
                              chained.Reject(e);
                          };
                          AddHandler(
                              resultHandler,
                              errorHandler,
                              chained.InternalCancel
                          );
                          return chained;
                      }
                      public IPromise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper) {
                          return Map(mapper, null);
                      }
                      /// <summary>
                      /// Сцепляет несколько аснхронных операций. Указанная асинхронная операция будет вызвана после
                      /// выполнения текущей, а результат текущей операции может быть использован для инициализации
                      /// новой операции.
                      /// </summary>
                      /// <typeparam name="TNew">Тип результата указанной асинхронной операции.</typeparam>
                      /// <param name="chained">Асинхронная операция, которая должна будет начаться после выполнения текущей.</param>
                      /// <param name="error">Обработчик ошибки. Данный обработчик получит
                      /// исключение возникшее при выполнении текуещй операции.</param>
                      /// <returns>Новое обещание, которое будет выполнено по окончанию указанной аснхронной операции.</returns>
                      public IPromise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained, ErrorHandler error) {
                          // проблема в том, что на момент связывания еще не начата асинхронная операция, поэтому нужно
                          // создать посредника, к которому будут подвызяваться следующие обработчики.
                          // когда будет выполнена реальная асинхронная операция, она обратиться к посреднику, чтобы
                          // передать через него результаты работы.
                          var medium = new Promise<TNew>(this, true);
                          ResultHandler<T> resultHandler = delegate(T result) {
                              if (medium.IsCancelled)
                                  return;
                              var promise = chained(result);
                              // notify chained operation that it's not needed
                              medium.Cancelled(() => promise.Cancel());
                              promise.Then(
                                  x => medium.Resolve(x),
                                  e => medium.Reject(e)
                              );
                          };
                          ErrorHandler errorHandler = delegate(Exception e) {
                              if (error != null)
                                  error(e);
                              // в случае ошибки нужно передать исключение дальше по цепочке
                              medium.Reject(e);
                          };
                          AddHandler(
                              resultHandler,
                              errorHandler,
                              medium.InternalCancel
                          );
                          return medium;
                      }
                      public IPromise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
                          return Chain(chained, null);
                      }
                      public IPromise<T> Cancelled(Action handler) {
                          AddHandler(null, null, handler);
                          return this;
                      }
                      /// <summary>
                      /// Adds the specified handler for all cases (success, error, cancel)
                      /// </summary>
                      /// <param name="handler">The handler that will be called anyway</param>
                      /// <returns>self</returns>
                      public IPromise<T> Finally(Action handler) {
                          if (handler == null)
                              throw new ArgumentNullException("handler");
                          AddHandler(
                              x => handler(),
                              e => handler(),
                              handler
                          );
                          return this;
                      }
                      /// <summary>
                      /// Дожидается отложенного обещания и в случае успеха, возвращает
                      /// его, результат, в противном случае бросает исключение.
                      /// </summary>
                      /// <remarks>
                      /// <para>
                      /// Если ожидание обещания было прервано по таймауту, это не значит,
                      /// что обещание было отменено или что-то в этом роде, это только
                      /// означает, что мы его не дождались, однако все зарегистрированные
                      /// обработчики, как были так остались и они будут вызваны, когда
                      /// обещание будет выполнено.
                      /// </para>
                      /// <para>
                      /// Такое поведение вполне оправдано поскольку таймаут может истечь
                      /// в тот момент, когда началась обработка цепочки обработчиков, и
                      /// к тому же текущее обещание может стоять в цепочке обещаний и его
                      /// отклонение может привести к непрогнозируемому результату.
                      /// </para>
                      /// </remarks>
                      /// <param name="timeout">Время ожидания</param>
                      /// <returns>Результат выполнения обещания</returns>
                      public T Join(int timeout) {
                          var evt = new ManualResetEvent(false);
                          Anyway(() => evt.Set());
                          Cancelled(() => evt.Set());
                          if (!evt.WaitOne(timeout, true))
                              throw new TimeoutException();
                          switch (m_state) {
                              case SucceededState:
                                  return m_result;
                              case CancelledState:
                                  throw new OperationCanceledException();
                              case RejectedState:
                                  throw new TargetInvocationException(m_error);
                              default:
                                  throw new ApplicationException(String.Format("Invalid promise state {0}", m_state));
                          }
                      }
                      public T Join() {
                          return Join(Timeout.Infinite);
                      }
                      void AddHandler(ResultHandler<T> success, ErrorHandler error, Action cancel) {
                          Interlocked.Increment(ref m_childrenCount);
                          HandlerDescriptor handler = new HandlerDescriptor {
                              resultHandler = success,
                              errorHandler = error,
                              cancellHandler = cancel
                          };
                          bool queued;
                          if (!IsResolved) {
                              m_handlers.Enqueue(handler);
                              queued = true;
                          } else {
                              // the promise is in resolved state, just invoke the handled with minimum overhead
                              queued = false;
                              InvokeHandler(handler);
                          }
                          if (queued && IsResolved && m_handlers.TryDequeue(out handler))
                              // if the promise have been resolved while we was adding handler to the queue
                              // we can't guarantee that someone is still processing it
                              // therefore we will fetch a handler from the queue and execute it
                              // note that fetched handler may be not the one that we have added
                              // even we can fetch no handlers at all :)
                              InvokeHandler(handler);
                      }
                      protected virtual void InvokeHandler(HandlerDescriptor handler) {
                          switch (m_state) {
                              case SucceededState:
                                  handler.Resolve(m_result);
                                  break;
                              case RejectedState:
                                  handler.Reject(m_error);
                                  break;
                              case CancelledState:
                                  handler.Cancel();
                                  break;
                              default:
                                  // do nothing
                                  return;
                          }
                      }
                      protected virtual void OnStateChanged() {
                          HandlerDescriptor handler;
                          while (m_handlers.TryDequeue(out handler))
                              InvokeHandler(handler);
                      }
                      public bool IsExclusive {
                          get {
                              return m_childrenCount <= 1;
                          }
                      }
                      protected bool Cancel(bool dependencies) {
                          if (BeginTransit()) {
                              CompleteTransit(CancelledState);
                              OnStateChanged();
                              if (dependencies && m_parent != null && m_parent.IsExclusive)
                                  m_parent.Cancel();
                              return true;
                          } else {
                              return false;
                          }
                      }
                      /// <summary>
                      /// Объединяет несколько обещаний в одно, результатом которого является массив результатов других обещаний.
                      /// Если хотябы одно из переданных обещаний не будет выполнено, то новое обещение тоже не будет выполнено.
                      /// При отмене нового обещания, переданные обещания также будут отменены, если никто больше на них не подписан.
                      /// </summary>
                      /// <param name="promises">Список обещаний. Если список пустой, то результирующее обещание возвращается уже выполненным.</param>
                      /// <returns>Обещание объединяющее в себе результат переданных обещаний.</returns>
                      /// <exception cref="ArgumentNullException"><paramref name="promises"/> не может быть null</exception>
                      public static Promise<T[]> CreateComposite(IList<Promise<T>> promises) {
                          if (promises == null)
                              throw new ArgumentNullException();
                          // создаем аккумулятор для результатов и результирующее обещание
                          var result = new T[promises.Count];
                          var promise = new Promise<T[]>();
                          // special case
                          if (promises.Count == 0) {
                              promise.Resolve(result);
                              return promise;
                          }
                          int pending = promises.Count;
                          for (int i = 0; i < promises.Count; i++) {
                              var dest = i;
                              promises[i].Then(
                                  x => {
                                      result[dest] = x;
                                      if(Interlocked.Decrement(ref pending) == 0)
                                          promise.Resolve(result);
                                  },
                                  e => promise.Reject(e)
                              );
                          }
                          promise.Cancelled(
                              () => {
                                  foreach(var d in promises)
                                      if(d.IsExclusive)
                                          d.Cancel();
                              }
                          );
                          return promise;
                      }
                      public static Promise<T> ResultToPromise(T result) {
                          var p = new Promise<T>();
                          p.Resolve(result);
                          return p;
                      }
                      public static Promise<T> ExceptionToPromise(Exception error) {
                          if (error == null)
                              throw new ArgumentNullException();
                          var p = new Promise<T>();
                          p.Reject(error);
                          return p;
                      }
                  }
              }

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages