implab/ImplabNet Commit - r105:4d308952fd5e

minor fixes

cin -

r105:4d308952fd5e v2

parent child

Context file:

r105:4d308952fd5e

Collapse all files

Implab.Test/AsyncTests.cs +15 0

              using System;
              using System.Reflection;
              using System.Threading;
              using Implab.Parallels;
              #if MONO
              using NUnit.Framework;
              using TestClassAttribute = NUnit.Framework.TestFixtureAttribute;
              using TestMethod = NUnit.Framework.TestAttribute;
              #else
              using Microsoft.VisualStudio.TestTools.UnitTesting;
              #endif
              namespace Implab.Test {
                  [TestClass]
                  public class AsyncTests {
                      [TestMethod]
                      public void ResolveTest() {
                          int res = -1;
                          var p = new Promise<int>();
                          p.Then(x => res = x);
                          p.Resolve(100);
                          Assert.AreEqual(100, res);
                      }
                      [TestMethod]
                      public void RejectTest() {
                          int res = -1;
                          Exception err = null;
                          var p = new Promise<int>();
                          p.Then(
                              x => res = x,
                              e => {
                                  err = e;
                                  return -2;
                              }
                          );
                          p.Reject(new ApplicationException("error"));
                          Assert.AreEqual(res, -1);
                          Assert.AreEqual(err.Message, "error");
                      }
                      [TestMethod]
                      public void CancelExceptionTest() {
                          var p = new Promise<bool>();
                          p.Cancel();
                          var p2 = p.Cancelled(() => {
                              throw new ApplicationException("CANCELLED");
                          });
                          try {
                              p2.Join();
                              Assert.Fail();
                          } catch (ApplicationException err) {
                              Assert.AreEqual("CANCELLED", err.InnerException.Message);
                          }
                      }
                      [TestMethod]
                      public void ContinueOnCancelTest() {
                          var p = new Promise<bool>();
                          p.Cancel();
                          var p2 = p
                              .Cancelled(() => {
                                  throw new ApplicationException("CANCELLED");
                              })
                              .Error(e => true);
                          Assert.AreEqual(true, p2.Join());
                      }
                      [TestMethod]
                      public void JoinSuccessTest() {
                          var p = new Promise<int>();
                          p.Resolve(100);
                          Assert.AreEqual(p.Join(), 100);
                      }
                      [TestMethod]
                      public void JoinFailTest() {
                          var p = new Promise<int>();
                          p.Reject(new ApplicationException("failed"));
                          try {
                              p.Join();
                              throw new ApplicationException("WRONG!");
                          } catch (TargetInvocationException err) {
                              Assert.AreEqual(err.InnerException.Message, "failed");
                          } catch {
                              Assert.Fail("Got wrong excaption");
                          }
                      }
                      [TestMethod]
                      public void MapTest() {
                          var p = new Promise<int>();
                          var p2 = p.Then(x => x.ToString());
                          p.Resolve(100);
                          Assert.AreEqual(p2.Join(), "100");
                      }
                      [TestMethod]
                      public void FixErrorTest() {
                          var p = new Promise<int>();
                          var p2 = p.Error(e => 101);
                          p.Reject(new Exception());
                          Assert.AreEqual(p2.Join(), 101);
                      }
                      [TestMethod]
                      public void ChainTest() {
                          var p1 = new Promise<int>();
                          var p3 = p1.Chain(x => {
                              var p2 = new Promise<string>();
                              p2.Resolve(x.ToString());
                              return p2;
                          });
                          p1.Resolve(100);
                          Assert.AreEqual(p3.Join(), "100");
                      }
                      [TestMethod]
+                     public void ChainFailTest() {
+                         var p1 = new Promise<int>();
+                         var p3 = p1.Chain(x => {
+                             var p2 = new Promise<string>();
+                             p2.Reject(new Exception("DIE!!!"));
+                             return p2;
+                         });
+                         p1.Resolve(100);
+                         Assert.IsTrue(p3.IsResolved);
+                     }
+                     [TestMethod]
                      public void PoolTest() {
                          var pid = Thread.CurrentThread.ManagedThreadId;
                          var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
                          Assert.AreNotEqual(pid, p.Join());
                      }
                      [TestMethod]
                      public void WorkerPoolSizeTest() {
                          var pool = new WorkerPool(5, 10, 1);
                          Assert.AreEqual(5, pool.PoolSize);
                          pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                          pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                          pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                          Assert.AreEqual(5, pool.PoolSize);
                          for (int i = 0; i < 100; i++)
                              pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                          Thread.Sleep(200);
                          Assert.AreEqual(10, pool.PoolSize);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void WorkerPoolCorrectTest() {
                          var pool = new WorkerPool(0,1000,100);
                          const int iterations = 1000;
                          int pending = iterations;
                          var stop = new ManualResetEvent(false);
                          var count = 0;
                          for (int i = 0; i < iterations; i++) {
                              pool
                                  .Invoke(() => 1)
                                  .Then(x => Interlocked.Add(ref count, x))
                                  .Then(x => Math.Log10(x))
                                  .Anyway(() => {
                                      Interlocked.Decrement(ref pending);
                                      if (pending == 0)
                                          stop.Set();
                                  });
                          }
                          stop.WaitOne();
                          Assert.AreEqual(iterations, count);
                          Console.WriteLine("Max threads: {0}", pool.MaxRunningThreads);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void WorkerPoolDisposeTest() {
                          var pool = new WorkerPool(5, 20);
                          Assert.AreEqual(5, pool.PoolSize);
                          pool.Dispose();
                          Thread.Sleep(500);
                          Assert.AreEqual(0, pool.PoolSize);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void MTQueueTest() {
                          var queue = new MTQueue<int>();
                          int res;
                          queue.Enqueue(10);
                          Assert.IsTrue(queue.TryDequeue(out res));
                          Assert.AreEqual(10, res);
                          Assert.IsFalse(queue.TryDequeue(out res));
                          for (int i = 0; i < 1000; i++)
                              queue.Enqueue(i);
                          for (int i = 0; i < 1000; i++) {
                              queue.TryDequeue(out res);
                              Assert.AreEqual(i, res);
                          }
                          int writers = 0;
                          int readers = 0;
                          var stop = new ManualResetEvent(false);
                          int total = 0;
                          const int itemsPerWriter = 10000;
                          const int writersCount = 10;
                          for (int i = 0; i < writersCount; i++) {
                              Interlocked.Increment(ref writers);
                              AsyncPool
                                  .InvokeNewThread(() => {
                                      for (int ii = 0; ii < itemsPerWriter; ii++) {
                                          queue.Enqueue(1);
                                      }
                                      return 1;
                                  })
                                  .Anyway(() => Interlocked.Decrement(ref writers));
                          }
                          for (int i = 0; i < 10; i++) {
                              Interlocked.Increment(ref readers);
                              AsyncPool
                                  .InvokeNewThread(() => {
                                      int t;
                                      do {
                                          while (queue.TryDequeue(out t))
                                              Interlocked.Add(ref total, t);
                                      } while (writers > 0);
                                      return 1;
                                  })
                                  .Anyway(() => {
                                      Interlocked.Decrement(ref readers);
                                      if (readers == 0)
                                          stop.Set();
                                  });
                          }
                          stop.WaitOne();
                          Assert.AreEqual(itemsPerWriter * writersCount, total);
                      }
                      [TestMethod]
                      public void ParallelMapTest() {
                          const int count = 100000;
                          var args = new double[count];
                          var rand = new Random();
                          for (int i = 0; i < count; i++)
                              args[i] = rand.NextDouble();
                          var t = Environment.TickCount;
                          var res = args.ParallelMap(x => Math.Sin(x*x), 4).Join();
                          Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                          t = Environment.TickCount;
                          for (int i = 0; i < count; i++)
                              Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                          Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                      }
                      [TestMethod]
                      public void ChainedMapTest() {
                          using (var pool = new WorkerPool(0,10,1)) {
                              const int count = 10000;
                              var args = new double[count];
                              var rand = new Random();
                              for (int i = 0; i < count; i++)
                                  args[i] = rand.NextDouble();
                              var t = Environment.TickCount;
                              var res = args
                                  .ChainedMap(
                                      // Analysis disable once AccessToDisposedClosure
                                      x => pool.Invoke(
                                          () => Math.Sin(x * x)
                                      ),
 
                                  )
                                  .Join();
                              Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                              t = Environment.TickCount;
                              for (int i = 0; i < count; i++)
                                  Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                              Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                              Console.WriteLine("Max workers: {0}", pool.MaxRunningThreads);
                          }
                      }
                      [TestMethod]
                      public void ParallelForEachTest() {
                          const int count = 100000;
                          var args = new int[count];
                          var rand = new Random();
                          for (int i = 0; i < count; i++)
                              args[i] = (int)(rand.NextDouble() * 100);
                          int result = 0;
                          var t = Environment.TickCount;
                          args.ParallelForEach(x => Interlocked.Add(ref result, x), 4).Join();
                          Console.WriteLine("Iteration complete in {0} ms, result: {1}", Environment.TickCount - t, result);
                          int result2 = 0;
                          t = Environment.TickCount;
                          for (int i = 0; i < count; i++)
                              result2 += args[i];
                          Assert.AreEqual(result2, result);
                          Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                      }
                      [TestMethod]
                      public void ComplexCase1Test() {
                          var flags = new bool[3];
                          // op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
                          var step1 = PromiseHelper
                              .Sleep(200, "Alan")
                              .Cancelled(() => flags[0] = true);
                          var p = step1
                              .Chain(x =>
                                  PromiseHelper
                                      .Sleep(200, "Hi, " + x)
                                      .Then(y => y)
                                      .Cancelled(() => flags[1] = true)
                              )
                              .Cancelled(() => flags[2] = true);
                          step1.Join();
                          p.Cancel();
                          try {
                              Assert.AreEqual(p.Join(), "Hi, Alan");
                              Assert.Fail("Shouldn't get here");
                          } catch (OperationCanceledException) {
                          }
                          Assert.IsFalse(flags[0]);
                          Assert.IsTrue(flags[1]);
                          Assert.IsTrue(flags[2]);
                      }
                      [TestMethod]
                      public void ChainedCancel1Test() {
                          // при отмене сцепленной асинхронной операции все обещание должно
                          // завершаться ошибкой OperationCanceledException
                          var p = PromiseHelper
                              .Sleep(1, "Hi, HAL!")
                              .Then(x => {
                                  // запускаем две асинхронные операции
                                  var result = PromiseHelper.Sleep(1000, "HEM ENABLED!!!");
                                  // вторая операция отменяет первую до завершения
                                  PromiseHelper
                                      .Sleep(100, "HAL, STOP!")
                                      .Then(result.Cancel);
                                  return result;
                              });
                          try {
                              p.Join();
                          } catch (TargetInvocationException err) {
                              Assert.IsTrue(err.InnerException is OperationCanceledException);
                          }
                      }
                      [TestMethod]
                      public void ChainedCancel2Test() {
                          // при отмене цепочки обещаний, вложенные операции также должны отменяться
                          var pSurvive = new Promise<bool>();
                          var hemStarted = new ManualResetEvent(false);
                          var p = PromiseHelper
                              .Sleep(1, "Hi, HAL!")
                              .Chain(x => {
                                  hemStarted.Set();
                                  // запускаем две асинхронные операции
                                  var result = PromiseHelper
                                      .Sleep(10000, "HEM ENABLED!!!")
                                      .Then(s => pSurvive.Resolve(false));
                                  result
                                      .Cancelled(() => pSurvive.Resolve(true));
                                  return result;
                              });
                          hemStarted.WaitOne();
                          p.Cancel();
                          try {
                              p.Join();
                          } catch (OperationCanceledException) {
                              Assert.IsTrue(pSurvive.Join());
                          }
                      }
                  }
              }

Implab/Promise.cs +1 -1

              using System;
              using System.Collections.Generic;
              using System.Reflection;
              using System.Threading;
              using Implab.Parallels;
              namespace Implab {
                  /// <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> {
                      protected struct HandlerDescriptor {
                          public Action<T> resultHandler;
                          public Func<Exception,T> errorHandler;
                          public Action cancellHandler;
                          public Promise<T> medium;
                          public void Resolve(T result) {
                              if (resultHandler != null) {
                                  try {
                                      resultHandler(result);
                                  } catch (Exception e) {
                                      Reject(e);
                                      return;
                                  }
                              }
                              if (medium != null)
                                  medium.Resolve(result);
                          }
                          public void Reject(Exception err) {
                              if (errorHandler != null) {
                                  try {
                                      var res = errorHandler(err);
                                      if (medium != null)
                                          medium.Resolve(res);
                                      /*} catch (TransientPromiseException err2) {
                                      if (medium != null)
                                          medium.Reject(err2.InnerException);*/
                                  } catch (Exception err2) {
                                      if (medium != null)
                                          medium.Reject(err2);
                                  }
                              } else if (medium != null)
                                  medium.Reject(err);
                          }
                          public void Cancel() {
                              if (cancellHandler != null) {
                                  try {
                                      cancellHandler();
                                  } catch (Exception err) {
                                      Reject(err);
                                      return;
                                  }
                              }
                              if (medium != null)
                                  medium.Cancel();
                          }
                      }
                      const int UNRESOLVED_SATE = 0;
                      const int TRANSITIONAL_STATE = 1;
                      const int SUCCEEDED_STATE = 2;
                      const int REJECTED_STATE = 3;
                      const int CANCELLED_STATE = 4;
                      int m_childrenCount;
                      int m_state;
                      T m_result;
                      Exception m_error;
                      readonly MTQueue<HandlerDescriptor> m_handlers = new MTQueue<HandlerDescriptor>();
                      public Promise() {
                      }
                      public Promise(IPromise parent) {
                          if (parent != null)
                              AddHandler(
                                  null,
                                  null,
                                  () => {
                                      if (parent.IsExclusive)
                                          parent.Cancel();
                                  },
                                  null,
                                  false
                              );
                      }
                      bool BeginTransit() {
                          return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE);
                      }
                      void CompleteTransit(int state) {
                          if (TRANSITIONAL_STATE != Interlocked.CompareExchange(ref m_state, state, TRANSITIONAL_STATE))
                              throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state");
                      }
                      void WaitTransition() {
                          while (m_state == TRANSITIONAL_STATE) {
                              Thread.MemoryBarrier();
                          }
                      }
                      public bool IsResolved {
                          get {
                              Thread.MemoryBarrier();
                              return m_state > 1;
                          }
                      }
                      public bool IsCancelled {
                          get {
                              Thread.MemoryBarrier();
                              return m_state == CANCELLED_STATE;
                          }
                      }
                      public Type PromiseType {
                          get { return typeof(T); }
                      }
                      /// <summary>
                      /// Выполняет обещание, сообщая об успешном выполнении.
                      /// </summary>
                      /// <param name="result">Результат выполнения.</param>
                      /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                      public void Resolve(T result) {
                          if (BeginTransit()) {
                              m_result = result;
                              CompleteTransit(SUCCEEDED_STATE);
                              OnStateChanged();
                          } else {
                              WaitTransition();
                              if (m_state != CANCELLED_STATE)
                                  throw new InvalidOperationException("The promise is already resolved");
                          }
                      }
                      /// <summary>
                      /// Выполняет обещание, сообщая об успешном выполнении. Результатом выполнения будет пустое значения.
                      /// </summary>
                      /// <remarks>
                      /// Данный вариант удобен в случаях, когда интересен факт выполнения операции, нежели полученное значение.
                      /// </remarks>
                      public void Resolve() {
                          Resolve(default(T));
                      }
                      /// <summary>
                      /// Выполняет обещание, сообщая об ошибке
                      /// </summary>
                      /// <remarks>
                      /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                      /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                      /// будут проигнорированы.
                      /// </remarks>
                      /// <param name="error">Исключение возникшее при выполнении операции</param>
                      /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                      public void Reject(Exception error) {
                          if (BeginTransit()) {
                              m_error = error is TransientPromiseException ? error.InnerException : error;
                              CompleteTransit(REJECTED_STATE);
                              OnStateChanged();
                          } else {
                              WaitTransition();
                              if (m_state == SUCCEEDED_STATE)
                                  throw new InvalidOperationException("The promise is already resolved");
                          }
                      }
                      /// <summary>
                      /// Отменяет операцию, если это возможно.
                      /// </summary>
                      /// <remarks>Для определения была ли операция отменена следует использовать свойство <see cref="IsCancelled"/>.</remarks>
                      public void Cancel() {
                          if (BeginTransit()) {
                              CompleteTransit(CANCELLED_STATE);
                              OnStateChanged();
                          }
                      }
                      public IPromise<T> Then(Action<T> success, Func<Exception,T> error, Action cancel) {
                          if (success == null && error == null && cancel == null)
                              return this;
                          var medium = new Promise<T>(this);
                          AddHandler(success, error, cancel, medium, true);
                          return medium;
                      }
                      /// <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(Action<T> success, Func<Exception,T> error) {
                          if (success == null && error == null)
                              return this;
                          var medium = new Promise<T>(this);
                          AddHandler(success, error, null, medium, true);
                          return medium;
                      }
                      public IPromise<T> Then(Action<T> success) {
                          if (success == null)
                              return this;
                          var medium = new Promise<T>(this);
                          AddHandler(success, null, null, medium, true);
                          return medium;
                      }
                      /// <summary>
                      /// Последний обработчик в цепочки обещаний.
                      /// </summary>
                      /// <param name="success"></param>
                      /// <param name="error"></param>
                      /// <param name="cancel"></param>
                      /// <remarks>
                      /// <para>
                      /// Данный метод не создает связанного с текущим обещания и предназначен для окончания
                      /// фсинхронной цепочки.
                      /// </para>
                      /// <para>
                      /// Если данный метод вызвать несколько раз, либо добавить другие обработчики, то цепочка
                      /// не будет одиночной <see cref="IsExclusive"/> и, как следствие, будет невозможна отмена
                      /// всей цепи обещаний снизу (с самого последнего обещания).
                      /// </para>
                      /// </remarks>
                      public void On(Action<T> success, Action<Exception> error, Action cancel) {
                          if (success == null && error == null && cancel == null)
                              return;
                          Func<Exception,T> errorHandler = null;
                          if (error != null)
                              errorHandler = err => {
                                  error(err);
                                  return default(T);
                              };
                          AddHandler(success, errorHandler, cancel, null, false);
                      }
                      public void On(Action<T> success, Action<Exception> error) {
                          On(success, error, null);
                      }
                      public void On(Action<T> success) {
                          On(success, null, null);
                      }
                      public void On(Action handler, PromiseEventType events) {
                          Safe.ArgumentNotNull(handler, "handler");
                          Action<T> success = events.HasFlag(PromiseEventType.Success) ? new Action<T>(x => handler()) : null;
                          Func<Exception,T> error = events.HasFlag(PromiseEventType.Error) ? new Func<Exception,T>(e => {
                              handler();
                              return default(T);
                          }) : null;
                          Action cancel = events.HasFlag(PromiseEventType.Cancelled) ? handler : null;
                          AddHandler(success, error, cancel, null, false);
                      }
                      public IPromise Error(Action<Exception> error) {
                          if (error == null)
                              return this;
                          var medium = new Promise<T>(this);
                          AddHandler(
                              null,
                              e => {
                                  error(e);
                                  return default(T);
                              },
                              null,
                              medium,
                              true
                          );
                          return medium;
                      }
                      /// <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(Func<Exception,T> handler) {
                          if (handler == null)
                              return this;
                          var medium = new Promise<T>(this);
                          AddHandler(null, handler, null, medium, true);
                          return medium;
                      }
                      /// <summary>
                      /// Позволяет преобразовать результат выполения операции к новому типу.
                      /// </summary>
                      /// <typeparam name="TNew">Новый тип результата.</typeparam>
                      /// <param name="mapper">Преобразование результата к новому типу.</param>
                      /// <param name="error">Обработчик ошибки. Данный обработчик получит
                      /// исключение возникшее при выполнении операции.</param>
                      /// <returns>Новое обещание, которое будет выполнено при выполнении исходного обещания.</returns>
                      /// <param name = "cancel"></param>
                      public IPromise<TNew> Then<TNew>(Func<T, TNew> mapper, Func<Exception,TNew> error, Action cancel) {
                          Safe.ArgumentNotNull(mapper, "mapper");
                          // создаем прицепленное обещание
                          var medium = new Promise<TNew>(this);
                          Action<T> resultHandler = result => medium.Resolve(mapper(result));
                          Func<Exception,T> errorHandler;
                          if (error != null)
                              errorHandler = e => {
                                  try {
                                      medium.Resolve(error(e));
                                  } catch (Exception e2) {
                                      // в случае ошибки нужно передать исключение дальше по цепочке
                                      medium.Reject(e2);
                                  }
                                  return default(T);
                              };
                          else
                              errorHandler = e => {
                                  medium.Reject(e);
                                  return default(T);
                              };
                          Action cancelHandler;
                          if (cancel != null)
                              cancelHandler = () => {
                                  cancel();
                                  medium.Cancel();
                              };
                          else
                              cancelHandler = medium.Cancel;
                          AddHandler(
                              resultHandler,
                              errorHandler,
                              cancelHandler,
                              null,
                              true
                          );
                          return medium;
                      }
                      public IPromise<TNew> Then<TNew>(Func<T, TNew> mapper, Func<Exception,TNew> error) {
                          return Then(mapper, error, null);
                      }
                      public IPromise<TNew> Then<TNew>(Func<T, TNew> mapper) {
                          return Then(mapper, null, null);
                      }
                      /// <summary>
                      /// Сцепляет несколько аснхронных операций. Указанная асинхронная операция будет вызвана после
                      /// выполнения текущей, а результат текущей операции может быть использован для инициализации
                      /// новой операции.
                      /// </summary>
                      /// <typeparam name="TNew">Тип результата указанной асинхронной операции.</typeparam>
                      /// <param name="chained">Асинхронная операция, которая должна будет начаться после выполнения текущей.</param>
                      /// <param name="error">Обработчик ошибки. Данный обработчик получит
                      /// исключение возникшее при выполнении текуещй операции.</param>
                      /// <returns>Новое обещание, которое будет выполнено по окончанию указанной аснхронной операции.</returns>
                      /// <param name = "cancel"></param>
                      public IPromise<TNew> Chain<TNew>(Func<T, IPromise<TNew>> chained, Func<Exception,IPromise<TNew>> error, Action cancel) {
                          Safe.ArgumentNotNull(chained, "chained");
                          // проблема в том, что на момент связывания еще не начата асинхронная операция, поэтому нужно
                          // создать посредника, к которому будут подвызяваться следующие обработчики.
                          // когда будет выполнена реальная асинхронная операция, она обратиться к посреднику, чтобы
                          // передать через него результаты работы.
                          var medium = new Promise<TNew>(this);
                          Action<T> resultHandler = delegate(T result) {
                              if (medium.IsCancelled)
                                  return;
                              var promise = chained(result);
                              promise.On(
                                  medium.Resolve,
                                  medium.Reject,
                                  () => medium.Reject(new OperationCanceledException()) // внешняя отмена связанной операции рассматривается как ошибка
                              );
                              // notify chained operation that it's not needed anymore
                              // порядок вызова Then, Cancelled важен, поскольку от этого
                              // зависит IsExclusive
                              medium.On(
                                  null,
                                  null,
                                  () => {
                                      if (promise.IsExclusive)
                                          promise.Cancel();
                                  }
                              );
                          };
                          Func<Exception,T> errorHandler;
                          if (error != null)
                              errorHandler = delegate(Exception e) {
                                  try {
                                      var promise = error(e);
                                      promise.On(
                                          medium.Resolve,
                                          medium.Reject,
                                          () => medium.Reject(new OperationCanceledException()) // внешняя отмена связанной операции рассматривается как ошибка
                                      );
                                      // notify chained operation that it's not needed anymore
                                      // порядок вызова Then, Cancelled важен, поскольку от этого
                                      // зависит IsExclusive
                                      medium.Cancelled(() => {
                                          if (promise.IsExclusive)
                                              promise.Cancel();
                                      });
                                  } catch (Exception e2) {
                                      medium.Reject(e2);
                                  }
                                  return default(T);
                              };
                          else
                              errorHandler = err => {
                                  medium.Reject(err);
                                  return default(T);
                              };
                          Action cancelHandler;
                          if (cancel != null)
                              cancelHandler = () => {
                                  if (cancel != null)
                                      cancel();
                                  medium.Cancel();
                              };
                          else
                              cancelHandler = medium.Cancel;
                          AddHandler(
                              resultHandler,
                              errorHandler,
                              cancelHandler,
                              null,
                              true
                          );
                          return medium;
                      }
                      public IPromise<TNew> Chain<TNew>(Func<T, IPromise<TNew>> chained, Func<Exception,IPromise<TNew>> error) {
                          return Chain(chained, error, null);
                      }
                      public IPromise<TNew> Chain<TNew>(Func<T, IPromise<TNew>> chained) {
                          return Chain(chained, null, null);
                      }
                      public IPromise<T> Cancelled(Action handler) {
                          var medium = new Promise<T>(this);
                          AddHandler(null, null, handler, medium, false);
                          return medium;
                      }
                      /// <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> Anyway(Action handler) {
                          Safe.ArgumentNotNull(handler, "handler");
                          var medium = new Promise<T>(this);
                          AddHandler(
                              x => handler(),
                              e => {
                                  handler();
                                  throw new TransientPromiseException(e);
                              },
                              handler,
                              medium,
                              true
                          );
                          return medium;
                      }
                      /// <summary>
                      /// Преобразует результат обещания к нужному типу
                      /// </summary>
                      /// <typeparam name="T2"></typeparam>
                      /// <returns></returns>
                      public IPromise<T2> Cast<T2>() {
                          return Then(x => (T2)(object)x, null);
                      }
                      /// <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());
                          if (!evt.WaitOne(timeout, true))
                              throw new TimeoutException();
                          switch (m_state) {
                              case SUCCEEDED_STATE:
                                  return m_result;
                              case CANCELLED_STATE:
                                  throw new OperationCanceledException();
                              case REJECTED_STATE:
                                  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(Action<T> success, Func<Exception,T> error, Action cancel, Promise<T> medium, bool inc) {
                          if (inc)
                              Interlocked.Increment(ref m_childrenCount);
                          var handler = new HandlerDescriptor {
                              resultHandler = success,
                              errorHandler = error,
                              cancellHandler = cancel,
                              medium = medium
                          };
                          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 SUCCEEDED_STATE:
                                  handler.Resolve(m_result);
                                  break;
                              case REJECTED_STATE:
                                  handler.Reject(m_error);
                                  break;
                              case CANCELLED_STATE:
                                  handler.Cancel();
                                  break;
                              default:
                                  // do nothing
                                  return;
                          }
                      }
                      void OnStateChanged() {
                          HandlerDescriptor handler;
                          while (m_handlers.TryDequeue(out handler))
                              InvokeHandler(handler);
                      }
                      public bool IsExclusive {
                          get {
                              return m_childrenCount <= 1;
                          }
                      }
                      /// <summary>
                      /// Объединяет несколько обещаний в одно, результатом которого является массив результатов других обещаний.
                      /// Если хотябы одно из переданных обещаний не будет выполнено, то новое обещение тоже не будет выполнено.
                      /// При отмене нового обещания, переданные обещания также будут отменены, если никто больше на них не подписан.
                      /// </summary>
                      /// <param name="promises">Список обещаний. Если список пустой, то результирующее обещание возвращается уже выполненным.</param>
                      /// <returns>Обещание объединяющее в себе результат переданных обещаний.</returns>
                      /// <exception cref="ArgumentNullException"><paramref name="promises"/> не может быть null</exception>
                      public static IPromise<T[]> CreateComposite(IList<IPromise<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;
                              if (promises[i] != null) {
                                  promises[i].Then(
                                      x => {
                                          result[dest] = x;
                                          if (Interlocked.Decrement(ref pending) == 0)
                                              promise.Resolve(result);
                                      },
                                      e => {
                                          promise.Reject(e);
                                          return default(T);
                                      }
                                  );
                              } else {
                                  if (Interlocked.Decrement(ref pending) == 0)
                                      promise.Resolve(result);
                              }
                          }
                          promise.Cancelled(
                              () => {
                                  foreach (var d in promises)
                                      if (d != null && d.IsExclusive)
                                          d.Cancel();
                              }
                          );
                          return promise;
                      }
                      /// <summary>
                      /// Объединяет несколько обещаний в одно. Результирующее обещание будет выполнено при
                      /// выполнении всех указанных обещаний. При этом возвращаемые значения первичных обещаний
                      /// игнорируются.
                      /// </summary>
                      /// <param name="promises">Коллекция первичных обещаний, которые будут объеденены в одно.</param>
                      /// <returns>Новое обещание, объединяющее в себе переданные.</returns>
                      /// <remarks>
                      /// Если в коллекции встречаюься <c>null</c>, то они воспринимаются как выполненные обещания.
                      /// </remarks>
                      public static IPromise CreateComposite(ICollection<IPromise> promises) {
                          if (promises == null)
                              throw new ArgumentNullException();
                          if (promises.Count == 0)
                              return Promise<object>.ResultToPromise(null);
                          int countdown = promises.Count;
                          var result = new Promise<object>();
                          foreach (var d in promises) {
                              if (d == null) {
                                  if (Interlocked.Decrement(ref countdown) == 0)
                                      result.Resolve(null);
                              } else {
                                  d.Then(() => {
                                      if (Interlocked.Decrement(ref countdown) == 0)
                                          result.Resolve(null);
                                  });
                              }
                          }
                          result.Cancelled(() => {
                              foreach (var d in promises)
                                  if (d != null && d.IsExclusive)
                                      d.Cancel();
                          });
                          return result;
                      }
                      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;
                      }
                      #region IPromiseBase explicit implementation
                      IPromise IPromise.Then(Action success, Action<Exception> error, Action cancel) {
                          return Then(
                              success != null ? new Action<T>(x => success()) : null,
                              error != null ? new Func<Exception,T>(e => {
                                  error(e);
                                  return default(T);
                              }) : null,
                              cancel
                          );
                      }
                      IPromise IPromise.Then(Action success, Action<Exception> error) {
                          return Then(
                              success != null ? new Action<T>(x => success()) : null,
                              error != null ? new Func<Exception,T>(e => {
                                  error(e);
                                  return default(T);
                              }) : null
                          );
                      }
                      IPromise IPromise.Then(Action success) {
                          Safe.ArgumentNotNull(success, "success");
                          return Then(x => success());
                      }
                      IPromise IPromise.Chain(Func<IPromise> chained, Func<Exception,IPromise> error, Action cancel) {
                          return ChainNoResult(chained, error, cancel);
                      }
                      IPromise ChainNoResult(Func<IPromise> chained, Func<Exception,IPromise> error, Action cancel) {
                          Safe.ArgumentNotNull(chained, "chained");
                          var medium = new Promise<object>(this);
                          Action<T> resultHandler = delegate {
                              if (medium.IsCancelled)
                                  return;
                              var promise = chained();
                              promise.On(
                                  medium.Resolve,
                                  medium.Reject,
                                  () => medium.Reject(new OperationCanceledException()) // внешняя отмена связанной операции рассматривается как ошибка
                              );
                              // notify chained operation that it's not needed anymore
                              // порядок вызова Then, Cancelled важен, поскольку от этого
                              // зависит IsExclusive
                              medium.Cancelled(() => {
                                  if (promise.IsExclusive)
                                      promise.Cancel();
                              });
                          };
                          Func<Exception,T> errorHandler;
                          if (error != null)
                              errorHandler = delegate(Exception e) {
                                  try {
                                      var promise = error(e);
                                      promise.On(
                                          medium.Resolve,
                                          medium.Reject,
                                          () => medium.Reject(new OperationCanceledException()) // внешняя отмена связанной операции рассматривается как ошибка
                                      );
                                      // notify chained operation that it's not needed anymore
                                      // порядок вызова Then, Cancelled важен, поскольку от этого
                                      // зависит IsExclusive
                                      medium.Cancelled(() => {
                                          if (promise.IsExclusive)
                                              promise.Cancel();
                                      });
                                  } catch (Exception e2) {
                                      medium.Reject(e2);
                                  }
                                  return default(T);
                              };
                          else
                              errorHandler = err => {
                                  medium.Reject(err);
                                  return default(T);
                              };
                          Action cancelHandler;
                          if (cancel != null)
                              cancelHandler = () => {
                                  if (cancel != null)
                                      cancel();
                                  medium.Cancel();
                              };
                          else
                              cancelHandler = medium.Cancel;
                          AddHandler(
                              resultHandler,
                              errorHandler,
                              cancelHandler,
                              null,
                              true
                          );
                          return medium;
                      }
                      IPromise IPromise.Chain(Func<IPromise> chained, Func<Exception,IPromise> error) {
                          return ChainNoResult(chained, error, null);
                      }
                      IPromise IPromise.Chain(Func<IPromise> chained) {
                          return ChainNoResult(chained, null, null);
                      }
                      void IPromise.On(Action success, Action<Exception> error, Action cancel) {
-                         On(x => success(), error, cancel);
+                         On(success != null ? new Action<T>(x => success()) : null, error, cancel);
                      }
                      void IPromise.On(Action success, Action<Exception> error) {
                          On(x => success(), error, null);
                      }
                      void IPromise.On(Action success) {
                          On(x => success(), null, null);
                      }
                      IPromise IPromise.Error(Action<Exception> error) {
                          return Error(error);
                      }
                      IPromise IPromise.Anyway(Action handler) {
                          return Anyway(handler);
                      }
                      IPromise IPromise.Cancelled(Action handler) {
                          return Cancelled(handler);
                      }
                      void IPromise.Join() {
                          Join();
                      }
                      void IPromise.Join(int timeout) {
                          Join(timeout);
                      }
                      #endregion
                  }
              }

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