Mercurial > pub > ImplabNet
view Implab/Promise.cs @ 107:f5220e5472ef v2
minor fixes and optimizations
author | cin |
---|---|
date | Tue, 11 Nov 2014 04:14:21 +0300 |
parents | d4e38929ce36 |
children | 38d6a4db35d7 |
line wrap: on
line source
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 abstract class AbstractHandler : MTCustomQueueNode<AbstractHandler> { public abstract void Resolve(T result); public abstract void Reject(Exception error); public abstract void Cancel(); } protected class RemapDescriptor<T2> : AbstractHandler { readonly Func<T,T2> m_resultHandler; readonly Func<Exception,T2> m_errorHandler; readonly Action m_cancellHandler; readonly Promise<T2> m_medium; public RemapDescriptor(Func<T,T2> resultHandler, Func<Exception,T2> errorHandler, Action cancelHandler, Promise<T2> medium) { m_resultHandler = resultHandler; m_errorHandler = errorHandler; m_cancellHandler = cancelHandler; m_medium = medium; } public override void Resolve(T result) { if (m_resultHandler != null) { try { if (m_medium != null) m_medium.Resolve(m_resultHandler(result)); else m_resultHandler(result); } catch (Exception e) { Reject(e); } } else if(m_medium != null) m_medium.Resolve(default(T2)); } public override void Reject(Exception error) { if (m_errorHandler != null) { try { var res = m_errorHandler(error); if (m_medium != null) m_medium.Resolve(res); } catch (Exception err2) { if (m_medium != null) m_medium.Reject(err2); } } else if (m_medium != null) m_medium.Reject(error); } public override void Cancel() { if (m_cancellHandler != null) { try { m_cancellHandler(); } catch (Exception err) { Reject(err); return; } } if (m_medium != null) m_medium.Cancel(); } } protected class HandlerDescriptor : AbstractHandler { readonly Action<T> m_resultHandler; readonly Action<Exception> m_errorHandler; readonly Action m_cancellHandler; readonly Promise<T> m_medium; public HandlerDescriptor(Action<T> resultHandler, Action<Exception> errorHandler, Action cancelHandler, Promise<T> medium) { m_resultHandler = resultHandler; m_errorHandler = errorHandler; m_cancellHandler = cancelHandler; m_medium = medium; } public override void Resolve(T result) { if (m_resultHandler != null) { try { m_resultHandler(result); } catch (Exception e) { Reject(e); return; } } if(m_medium != null) m_medium.Resolve(result); } public override void Reject(Exception error) { if (m_errorHandler != null) { try { m_errorHandler(error); if (m_medium != null) m_medium.Resolve(default(T)); } catch (Exception err2) { if (m_medium != null) m_medium.Reject(err2); } } else if (m_medium != null) m_medium.Reject(error); } public override void Cancel() { if (m_cancellHandler != null) { try { m_cancellHandler(); } catch (Exception err) { Reject(err); return; } } if (m_medium != null) m_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 MTCustomQueue<AbstractHandler> m_handlers = new MTCustomQueue<AbstractHandler>(); //readonly MTQueue<AbstractHandler> m_handlers = new MTQueue<AbstractHandler>(); public Promise() { } public Promise(IPromise parent) { if (parent != null) AddMappers<T>( 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(); } } /// <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; AddHandler(success, error, cancel, null, false); } public void On(Action<T> success, Action<Exception> error) { AddHandler(success, error, null, null, false); } public void On(Action<T> success) { AddHandler(success, null, null, null, false); } public void On(Action handler, PromiseEventType events) { Safe.ArgumentNotNull(handler, "handler"); AddHandler( events.HasFlag(PromiseEventType.Success) ? new Action<T>(x => handler()) : null, events.HasFlag(PromiseEventType.Error) ? new Action<Exception>( x => handler()) : null, events.HasFlag(PromiseEventType.Cancelled) ? handler : null, null, false ); } public IPromise Error(Action<Exception> error) { if (error == null) return this; var medium = new Promise<T>(this); AddMappers( 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); AddMappers(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); AddMappers( mapper, error, cancel, medium, 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); Func<T,T> resultHandler = delegate(T result) { if (medium.IsCancelled) return default(T); 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(); } ); return default(T); }; 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; AddMappers( 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 AddMappers<T2>(Func<T,T2> success, Func<Exception,T2> error, Action cancel, Promise<T2> medium, bool inc) { if (inc) Interlocked.Increment(ref m_childrenCount); AbstractHandler handler = new RemapDescriptor<T2>(success, error, cancel, 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); } void AddHandler(Action<T> success, Action<Exception> error, Action cancel, Promise<T> medium, bool inc) { if (inc) Interlocked.Increment(ref m_childrenCount); AbstractHandler handler = new HandlerDescriptor(success, error, cancel, 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(AbstractHandler 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() { AbstractHandler 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].On( x => { result[dest] = x; if (Interlocked.Decrement(ref pending) == 0) promise.Resolve(result); }, promise.Reject ); } 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 Func<T,T>(x => { success(); return x; }) : 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 Func<T,T>(x => { success(); return x; }) : 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(); return x; }); } 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); Func<T,T> resultHandler = delegate { if (medium.IsCancelled) return default(T); 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(); }); return default(T); }; 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; AddMappers( 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(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 } }