|
|
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 {
|
|
|
|
|
|
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 ResolvedState = 2;
|
|
|
const int RejectedState = 3;
|
|
|
const int CancelledState = 4;
|
|
|
|
|
|
readonly IPromise 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(IPromise 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");
|
|
|
}
|
|
|
|
|
|
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(ResolvedState);
|
|
|
OnStateChanged();
|
|
|
} else 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 if (m_state == ResolvedState)
|
|
|
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 Promise<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;
|
|
|
}
|
|
|
|
|
|
/// <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 Promise<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 { }
|
|
|
medium.Reject(x);
|
|
|
};
|
|
|
else
|
|
|
errorHandler = medium.Reject;
|
|
|
|
|
|
AddHandler(resultHandler, errorHandler, medium.InternalCancel);
|
|
|
|
|
|
return medium;
|
|
|
}
|
|
|
|
|
|
|
|
|
public Promise<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 Promise<T> Error(ErrorHandler error) {
|
|
|
return Then(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 Promise<T> Error(ErrorHandler<T> handler) {
|
|
|
if (handler == null)
|
|
|
return this;
|
|
|
|
|
|
var medium = new Promise<T>(this, true);
|
|
|
|
|
|
AddHandler(
|
|
|
null,
|
|
|
e => {
|
|
|
try {
|
|
|
medium.Resolve(handler(e));
|
|
|
} catch (Exception e2) {
|
|
|
medium.Reject(e2);
|
|
|
}
|
|
|
},
|
|
|
medium.InternalCancel
|
|
|
);
|
|
|
|
|
|
return medium;
|
|
|
}
|
|
|
|
|
|
public Promise<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 Promise<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 Promise<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 Promise<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 Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
|
|
|
return Chain(chained, null);
|
|
|
}
|
|
|
|
|
|
public Promise<T> Cancelled(Action handler) {
|
|
|
AddHandler(null, null, handler);
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
public Promise<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 ResolvedState:
|
|
|
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 we have added
|
|
|
InvokeHandler(handler);
|
|
|
|
|
|
}
|
|
|
|
|
|
void InvokeHandler(HandlerDescriptor handler) {
|
|
|
switch (m_state) {
|
|
|
case ResolvedState:
|
|
|
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;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
}
|
|
|
}
|
|
|
|
