##// END OF EJS Templates
implab » ImplabNet
RSS
Improved worker pool
Improved worker pool
cin - - Load All Authors

File last commit:

r16:5a4b735ba669 promises
r17:7cd4a843b4e4 promises
Show More
Promise.cs
549 lines | 23.7 KiB | text/x-csharp | CSharpLexer
/ Implab / Promise.cs
History | Source | Raw |Copy content |Copy permalink
cin
small fixes
 r2 using System;
using System.Collections.Generic;
using System.Reflection;
using System.Diagnostics;
using System.Threading;
namespace Implab {
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public delegate void ErrorHandler(Exception e);
cin
refactoring
 r11 public delegate T ErrorHandler<out T>(Exception e);
cin
implemeted new cancellable promises concept
 r10 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);
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <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>
cin
sync
 r9 public class Promise<T> : IPromise {
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 struct ResultHandlerInfo {
public ResultHandler<T> resultHandler;
public ErrorHandler errorHandler;
}
cin
implemeted new cancellable promises concept
 r10 readonly IPromise m_parent;
cin
sync
 r9
LinkedList<ResultHandlerInfo> m_resultHandlers = new LinkedList<ResultHandlerInfo>();
LinkedList<Action> m_cancelHandlers = new LinkedList<Action>();
cin
implemeted new cancellable promises concept
 r10 readonly object m_lock = new Object();
readonly bool m_cancellable;
int m_childrenCount = 0;
cin
sync
 r9
cin
inital progress handling
 r7 PromiseState m_state;
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 T m_result;
Exception m_error;
cin
sync
 r9
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public Promise() {
m_cancellable = true;
}
cin
small fixes
 r2
cin
sync
 r9 public Promise(IPromise parent, bool cancellable) {
cin
inital progress handling
 r7 m_cancellable = cancellable;
m_parent = parent;
cin
implemeted new cancellable promises concept
 r10 if (parent != null)
parent.HandleCancelled(InternalCancel);
cin
inital progress handling
 r7 }
cin
implemeted new cancellable promises concept
 r10 void InternalCancel() {
// don't try to cancel parent :)
Cancel(false);
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <summary>
/// Выполняет обещание, сообщая об успешном выполнении.
/// </summary>
/// <param name="result">Результат выполнения.</param>
/// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
public void Resolve(T result) {
cin
Implemented interllocked queue...
 r14 lock (m_lock) {
cin
inital progress handling
 r7 if (m_state == PromiseState.Cancelled)
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return;
cin
inital progress handling
 r7 if (m_state != PromiseState.Unresolved)
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 throw new InvalidOperationException("The promise is already resolved");
m_result = result;
cin
inital progress handling
 r7 m_state = PromiseState.Resolved;
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 OnStateChanged();
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <summary>
/// Выполняет обещание, сообщая об ошибке
/// </summary>
cin
sync
 r16 /// <remarks>
/// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
/// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
/// будут проигнорированы.
/// </remarks>
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <param name="error">Исключение возникшее при выполнении операции</param>
/// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
public void Reject(Exception error) {
cin
Implemented interllocked queue...
 r14 lock (m_lock) {
cin
sync
 r16 if (m_state == PromiseState.Cancelled || m_state == PromiseState.Rejected)
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return;
cin
inital progress handling
 r7 if (m_state != PromiseState.Unresolved)
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 throw new InvalidOperationException("The promise is already resolved");
m_error = error;
cin
inital progress handling
 r7 m_state = PromiseState.Rejected;
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 OnStateChanged();
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <summary>
/// Отменяет операцию, если это возможно.
/// </summary>
/// <returns><c>true</c> Операция была отменена, обработчики не будут вызваны.<c>false</c> отмена не возможна, поскольку обещание уже выполнено и обработчики отработали.</returns>
public bool Cancel() {
cin
sync
 r9 return Cancel(true);
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <summary>
cin
refactoring
 r11 /// Adds new handlers to this promise.
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// </summary>
cin
refactoring
 r11 /// <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>
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public Promise<T> Then(ResultHandler<T> success, ErrorHandler error) {
if (success == null && error == null)
return this;
cin
refactoring
 r11 var medium = new Promise<T>(this, true);
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6
var handlerInfo = new ResultHandlerInfo();
if (success != null)
handlerInfo.resultHandler = x => {
cin
implemeted new cancellable promises concept
 r10 success(x);
medium.Resolve(x);
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 };
else
cin
implemeted new cancellable promises concept
 r10 handlerInfo.resultHandler = medium.Resolve;
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 if (error != null)
handlerInfo.errorHandler = x => {
try {
error(x);
} catch { }
medium.Reject(x);
};
else
cin
implemeted new cancellable promises concept
 r10 handlerInfo.errorHandler = medium.Reject;
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 AddHandler(handlerInfo);
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return medium;
}
cin
small fixes
 r2
cin
refactoring
 r11 /// <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);
var handlerInfo = new ResultHandlerInfo();
if (success != null)
handlerInfo.resultHandler = x => {
success(x);
medium.Resolve(x);
};
else
handlerInfo.resultHandler = medium.Resolve;
if (error != null)
handlerInfo.errorHandler = x => {
try {
medium.Resolve(error(x));
} catch { }
medium.Reject(x);
};
else
handlerInfo.errorHandler = medium.Reject;
AddHandler(handlerInfo);
return medium;
}
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public Promise<T> Then(ResultHandler<T> success) {
cin
refactoring
 r11 if (success == null)
return this;
var medium = new Promise<T>(this, true);
var handlerInfo = new ResultHandlerInfo();
if (success != null)
handlerInfo.resultHandler = x => {
success(x);
medium.Resolve(x);
};
else
handlerInfo.resultHandler = medium.Resolve;
handlerInfo.errorHandler = medium.Reject;
AddHandler(handlerInfo);
return medium;
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public Promise<T> Error(ErrorHandler error) {
return Then(null, error);
}
cin
small fixes
 r2
cin
refactoring
 r11 /// <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(new ResultHandlerInfo {
errorHandler = e => {
try {
medium.Resolve(handler(e));
} catch (Exception e2) {
medium.Reject(e2);
}
}
});
return medium;
}
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 public Promise<T> Anyway(Action handler) {
if (handler == null)
return this;
var medium = new Promise<T>();
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 AddHandler(new ResultHandlerInfo {
resultHandler = x => {
cin
implemeted new cancellable promises concept
 r10 // to avoid handler being called multiple times we handle exception by ourselfs
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 try {
handler();
medium.Resolve(x);
} catch (Exception e) {
medium.Reject(e);
}
},
errorHandler = x => {
try {
handler();
} catch { }
medium.Reject(x);
}
});
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return medium;
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <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");
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 // создаем прицепленное обещание
cin
implemeted new cancellable promises concept
 r10 var chained = new Promise<TNew>();
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 AddHandler(new ResultHandlerInfo() {
cin
implemeted new cancellable promises concept
 r10 resultHandler = result => chained.Resolve(mapper(result)),
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 errorHandler = delegate(Exception e) {
if (error != null)
cin
implemeted new cancellable promises concept
 r10 try {
error(e);
} catch { }
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 // в случае ошибки нужно передать исключение дальше по цепочке
chained.Reject(e);
}
});
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return chained;
}
public Promise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper) {
return Map(mapper, null);
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <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) {
// проблема в том, что на момент связывания еще не начата асинхронная операция, поэтому нужно
// создать посредника, к которому будут подвызяваться следующие обработчики.
// когда будет выполнена реальная асинхронная операция, она обратиться к посреднику, чтобы
// передать через него результаты работы.
cin
implemeted new cancellable promises concept
 r10 var medium = new Promise<TNew>(this, true);
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 AddHandler(new ResultHandlerInfo {
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 resultHandler = delegate(T result) {
cin
implemeted new cancellable promises concept
 r10 if (medium.State == PromiseState.Cancelled)
return;
var promise = chained(result);
// notify chained operation that it's not needed
medium.Cancelled(() => promise.Cancel());
promise.Then(
cin
refactoring
 r11 x => medium.Resolve(x),
e => medium.Reject(e)
);
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 },
errorHandler = delegate(Exception e) {
if (error != null)
error(e);
// в случае ошибки нужно передать исключение дальше по цепочке
medium.Reject(e);
}
});
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 return medium;
}
public Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
return Chain(chained, null);
}
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 public Promise<T> Cancelled(Action handler) {
if (handler == null)
return this;
lock (m_lock) {
if (m_state == PromiseState.Unresolved)
m_cancelHandlers.AddLast(handler);
else if (m_state == PromiseState.Cancelled)
handler();
}
return this;
}
public void HandleCancelled(Action handler) {
Cancelled(handler);
}
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 /// <summary>
/// Дожидается отложенного обещания и в случае успеха, возвращает
/// его, результат, в противном случае бросает исключение.
/// </summary>
/// <remarks>
/// <para>
/// Если ожидание обещания было прервано по таймауту, это не значит,
/// что обещание было отменено или что-то в этом роде, это только
/// означает, что мы его не дождались, однако все зарегистрированные
/// обработчики, как были так остались и они будут вызваны, когда
/// обещание будет выполнено.
/// </para>
/// <para>
/// Такое поведение вполне оправдано поскольку таймаут может истечь
/// в тот момент, когда началась обработка цепочки обработчиков, и
/// к тому же текущее обещание может стоять в цепочке обещаний и его
/// отклонение может привести к непрогнозируемому результату.
/// </para>
/// </remarks>
/// <param name="timeout">Время ожидания</param>
/// <returns>Результат выполнения обещания</returns>
public T Join(int timeout) {
cin
implemeted new cancellable promises concept
 r10 var evt = new ManualResetEvent(false);
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 Anyway(() => evt.Set());
cin
implemeted new cancellable promises concept
 r10 Cancelled(() => evt.Set());
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 if (!evt.WaitOne(timeout, true))
throw new TimeoutException();
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 switch (State) {
case PromiseState.Resolved:
return m_result;
case PromiseState.Cancelled:
throw new OperationCanceledException();
case PromiseState.Rejected:
throw new TargetInvocationException(m_error);
default:
throw new ApplicationException(String.Format("Invalid promise state {0}", State));
}
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
public T Join() {
return Join(Timeout.Infinite);
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 void AddHandler(ResultHandlerInfo handler) {
bool invokeRequired = false;
cin
small fixes
 r2
cin
implemeted new cancellable promises concept
 r10 lock (m_lock) {
m_childrenCount++;
if (m_state == PromiseState.Unresolved) {
cin
sync
 r9 m_resultHandlers.AddLast(handler);
cin
implemeted new cancellable promises concept
 r10 } else
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 invokeRequired = true;
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 // обработчики не должны блокировать сам объект
if (invokeRequired)
InvokeHandler(handler);
}
cin
small fixes
 r2
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 void InvokeHandler(ResultHandlerInfo handler) {
cin
implemeted new cancellable promises concept
 r10 switch (m_state) {
case PromiseState.Resolved:
try {
if (handler.resultHandler != null)
handler.resultHandler(m_result);
} catch (Exception e) {
try {
if (handler.errorHandler != null)
handler.errorHandler(e);
} catch { }
}
break;
case PromiseState.Rejected:
try {
if (handler.errorHandler != null)
handler.errorHandler(m_error);
} catch { }
break;
default:
// do nothing
return;
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
}
cin
small fixes
 r2
cin
refactoring
 r11 protected virtual void OnStateChanged() {
switch (m_state) {
case PromiseState.Resolved:
foreach (var resultHandlerInfo in m_resultHandlers)
try {
if (resultHandlerInfo.resultHandler != null)
resultHandlerInfo.resultHandler(m_result);
} catch (Exception e) {
try {
if (resultHandlerInfo.errorHandler != null)
resultHandlerInfo.errorHandler(e);
} catch { }
}
break;
case PromiseState.Cancelled:
foreach (var cancelHandler in m_cancelHandlers)
cancelHandler();
break;
case PromiseState.Rejected:
foreach (var resultHandlerInfo in m_resultHandlers)
try {
if (resultHandlerInfo.errorHandler != null)
resultHandlerInfo.errorHandler(m_error);
} catch { }
break;
default:
throw new InvalidOperationException(String.Format("Promise entered an invalid state {0}", m_state));
}
m_resultHandlers = null;
m_cancelHandlers = null;
}
cin
small fixes
 r2
cin
sync
 r9
public bool IsExclusive {
get {
lock (m_lock) {
return m_childrenCount <= 1;
}
}
}
public PromiseState State {
get {
lock (m_lock) {
return m_state;
}
}
}
cin
refactoring
 r11 protected bool Cancel(bool dependencies) {
cin
sync
 r9 bool result;
lock (m_lock) {
if (m_state == PromiseState.Unresolved) {
m_state = PromiseState.Cancelled;
result = true;
} else {
result = false;
}
}
cin
implemeted new cancellable promises concept
 r10 if (result)
OnStateChanged();
cin
sync
 r9
cin
implemeted new cancellable promises concept
 r10 if (dependencies && m_parent != null && m_parent.IsExclusive) {
cin
refactoring
 r11 m_parent.Cancel();
cin
sync
 r9 }
return result;
}
cin
implemented parallel map and foreach for arrays...
 r15
cin
*refactoring: Promise.Then now returns a new chained promise...
 r6 }
cin
small fixes
 r2 }