using System; using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; namespace Tapeti.Saga { public class SagaProvider : ISagaProvider { protected static readonly ConcurrentDictionary SagaLocks = new ConcurrentDictionary(); private readonly ISagaStore store; public SagaProvider(ISagaStore store) { this.store = store; } public async Task> Begin(T initialState) where T : class { var saga = await Saga.Create(() => Task.FromResult(initialState)); await store.Update(saga.Id, saga.State); return saga; } public async Task> Continue(string sagaId) where T : class { return await Saga.Create(async () => await store.Read(sagaId) as T, sagaId); } protected class Saga : ISaga where T : class { private bool disposed; public string Id { get; set; } public T State { get; set; } public static async Task> Create(Func> getState, string id = null) { var sagaId = id ?? Guid.NewGuid().ToString(); await SagaLocks.GetOrAdd(sagaId, new SemaphoreSlim(1)).WaitAsync(); var saga = new Saga { Id = sagaId, State = await getState() }; return saga; } public void Dispose() { if (disposed) return; SemaphoreSlim semaphore; if (SagaLocks.TryGetValue(Id, out semaphore)) semaphore.Release(); disposed = true; } public void ExpectResponse(string callId) { throw new NotImplementedException(); } public void ResolveResponse(string callId) { throw new NotImplementedException(); } } } }